remove old queues
This commit is contained in:
Alex Holliday committed
1 parent
47162f2064
commit
d2461ef118
11 files changed
+16
-1089
No files matched your search
Generated
-113
@@ -31,7 +31,6 @@
|
||||
"jmespath": "^0.16.0",
|
||||
"jsdom": "^26.1.0",
|
||||
"jsonwebtoken": "9.0.2",
|
||||
"less-simple-scheduler": "2.6.0",
|
||||
"mailersend": "^2.2.0",
|
||||
"mjml": "^5.0.0-alpha.4",
|
||||
"mongoose": "^8.3.3",
|
||||
@@ -40,7 +39,6 @@
|
||||
"ping": "0.4.4",
|
||||
"sharp": "0.33.5",
|
||||
"ssl-checker": "2.0.10",
|
||||
"super-simple-scheduler": "1.9.2",
|
||||
"swagger-ui-express": "5.0.1",
|
||||
"winston": "^3.13.0",
|
||||
"ws": "^8.19.0",
|
||||
@@ -10608,95 +10606,6 @@
|
||||
"integrity": "sha512-Xq9nH7KlWZmXAtodXDDRE7vs6DU1gTU8zYDHDiWLSip45Egwq3plLHzPn27NgvzL2r1LMPC1vdqh98sQxtqj4A==",
|
||||
"license": "MIT"
|
||||
},
|
||||
"node_modules/less-simple-scheduler": {
|
||||
"version": "2.6.0",
|
||||
"resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.6.0.tgz",
|
||||
"integrity": "sha512-Vi8ouiodLqe2fyh+DqCmxE5cCJZDI3xOyUAH4jcj9sf7GunLUzH71PBcYJy2P8lxTK3LxEkHtr1reymiogs6JQ==",
|
||||
"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",
|
||||
@@ -14754,28 +14663,6 @@
|
||||
"postcss": "^8.4.32"
|
||||
}
|
||||
},
|
||||
"node_modules/super-simple-scheduler": {
|
||||
"version": "1.9.2",
|
||||
"resolved": "https://registry.npmjs.org/super-simple-scheduler/-/super-simple-scheduler-1.9.2.tgz",
|
||||
"integrity": "sha512-dfeI1R8jFLeHXIrNsDRxqvdJN2r4BjwKrvv/pPThu0X4RQa008/ubkFno96esHad04J6KyWYTK+P/qgOcoaReg==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"uuid": "11.1.1"
|
||||
}
|
||||
},
|
||||
"node_modules/super-simple-scheduler/node_modules/uuid": {
|
||||
"version": "11.1.1",
|
||||
"resolved": "https://registry.npmjs.org/uuid/-/uuid-11.1.1.tgz",
|
||||
"integrity": "sha512-vIYxrBCC/N/K+Js3qSN88go7kIfNPssr/hHCesKCQNAjmgvYS2oqr69kIufEG+O4+PfezOH4EbIeHCfFov8ZgQ==",
|
||||
"funding": [
|
||||
"https://github.com/sponsors/broofa",
|
||||
"https://github.com/sponsors/ctavan"
|
||||
],
|
||||
"license": "MIT",
|
||||
"bin": {
|
||||
"uuid": "dist/esm/bin/uuid"
|
||||
}
|
||||
},
|
||||
"node_modules/supports-color": {
|
||||
"version": "7.2.0",
|
||||
"resolved": "https://registry.npmjs.org/supports-color/-/supports-color-7.2.0.tgz",
|
||||
|
||||
@@ -53,7 +53,6 @@
|
||||
"jmespath": "^0.16.0",
|
||||
"jsdom": "^26.1.0",
|
||||
"jsonwebtoken": "9.0.2",
|
||||
"less-simple-scheduler": "2.6.0",
|
||||
"mailersend": "^2.2.0",
|
||||
"mjml": "^5.0.0-alpha.4",
|
||||
"mongoose": "^8.3.3",
|
||||
@@ -62,7 +61,6 @@
|
||||
"ping": "0.4.4",
|
||||
"sharp": "0.33.5",
|
||||
"ssl-checker": "2.0.10",
|
||||
"super-simple-scheduler": "1.9.2",
|
||||
"swagger-ui-express": "5.0.1",
|
||||
"winston": "^3.13.0",
|
||||
"ws": "^8.19.0",
|
||||
|
||||
@@ -30,8 +30,6 @@ import { EmailService, IEmailService } from "@/service/emailService.js";
|
||||
import { GlobalPingService } from "@/service/globalPingService.js";
|
||||
import { WorkerHelper } from "@/worker/worker.helper.js";
|
||||
import { IWorker } from "@/worker/worker.interface.js";
|
||||
import { LessSimpleWorker } from "@/worker/worker.less-simple.js";
|
||||
import { SuperSimpleQueue } from "@/worker/worker.super-simple.js";
|
||||
import { INetworkService, NetworkService } from "@/service/networkService.js";
|
||||
import { IStatusService, StatusService } from "@/service/statusService.js";
|
||||
import { MonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js";
|
||||
@@ -89,7 +87,6 @@ import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interf
|
||||
import MongoMonitorsRepository from "@/domain/monitors/monitor.repository.mongo.js";
|
||||
import { INotificationsRepository } from "@/domain/notifications/notification.repository.interface.js";
|
||||
import MongoNotificationsRepository from "@/domain/notifications/notification.repository.mongo.js";
|
||||
import MongoQueueWorkersRepository from "@/domain/queue-workers/queue-worker.repository.mongo.js";
|
||||
import { IRecoveryTokensRepository } from "@/domain/recovery-tokens/recovery-token.repository.interface.js";
|
||||
import MongoRecoveryTokensRepository from "@/domain/recovery-tokens/recovery-token.repository.mongo.js";
|
||||
import { IStatusPagesRepository } from "@/domain/status-pages/status-page-repository.interface.js";
|
||||
@@ -101,7 +98,6 @@ import MongoTeamsRepository from "@/domain/teams/team.repository.model.js";
|
||||
import { IUsersRepository } from "@/domain/users/user.repository.interface.js";
|
||||
import MongoUsersRepository from "@/domain/users/user.repository.mongo.js";
|
||||
import { ILogger } from "@/utils/logger.js";
|
||||
import { AppError } from "@/utils/AppError.js";
|
||||
import { type QueueMode } from "@/domain/app-settings/app-settings.type.js";
|
||||
import { NotificationReactor } from "@/worker/reactors/reactor.notification.js";
|
||||
import { ReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js";
|
||||
@@ -155,12 +151,10 @@ export const initializeServices = async ({
|
||||
logger,
|
||||
envSettings,
|
||||
settingsService,
|
||||
queueMode = "primary",
|
||||
}: {
|
||||
logger: ILogger;
|
||||
envSettings: EnvConfig;
|
||||
settingsService: ISettingsService;
|
||||
queueMode?: QueueMode;
|
||||
}): Promise<InitializedServices> => {
|
||||
// Create DB
|
||||
|
||||
@@ -168,58 +162,6 @@ export const initializeServices = async ({
|
||||
db = new MongoDB(logger, envSettings);
|
||||
await db.connect();
|
||||
|
||||
// ** NOTE **
|
||||
// const dbType = envSettings.dbType;
|
||||
// DB type has been fixed to MongoDB for now
|
||||
|
||||
// if (dbType === "mongodb") {
|
||||
// db = new MongoDB(logger, envSettings);
|
||||
// }
|
||||
|
||||
// if (!db) {
|
||||
// throw new AppError({ message: "Unsupported database type", status: 500 });
|
||||
// }
|
||||
|
||||
// await db.connect();
|
||||
|
||||
// Repositories
|
||||
|
||||
// ** NOTE **
|
||||
// DB Type fixed to MongoDB for now
|
||||
// let monitorsRepository: IMonitorsRepository;
|
||||
// let checksRepository: IChecksRepository;
|
||||
// let geoChecksRepository: IGeoChecksRepository;
|
||||
// let monitorStatsRepository: IMonitorStatsRepository;
|
||||
// let statusPagesRepository: IStatusPagesRepository;
|
||||
// let usersRepository: IUsersRepository;
|
||||
// let invitesRepository: IInvitesRepository;
|
||||
// let recoveryTokensRepository: IRecoveryTokensRepository;
|
||||
// let settingsRepository: ISettingsRepository;
|
||||
// let notificationsRepository: INotificationsRepository;
|
||||
// let tagsRepository: ITagsRepository;
|
||||
// let incidentsRepository: IIncidentsRepository;
|
||||
// let teamsRepository: ITeamsRepository;
|
||||
// let maintenanceWindowsRepository: IMaintenanceWindowsRepository;
|
||||
|
||||
// if (dbType === "mongodb") {
|
||||
// monitorsRepository = new MongoMonitorsRepository();
|
||||
// checksRepository = new MongoChecksRepository(logger);
|
||||
// geoChecksRepository = new MongoGeoChecksRepository(logger);
|
||||
// monitorStatsRepository = new MongoMonitorStatsRepository();
|
||||
// statusPagesRepository = new MongoStatusPagesRepository();
|
||||
// usersRepository = new MongoUsersRepository();
|
||||
// invitesRepository = new MongoInvitesRepository();
|
||||
// recoveryTokensRepository = new MongoRecoveryTokensRepository();
|
||||
// settingsRepository = new MongoSettingsRepository();
|
||||
// notificationsRepository = new MongoNotificationsRepository();
|
||||
// tagsRepository = new MongoTagsRepository();
|
||||
// incidentsRepository = new MongoIncidentsRepository();
|
||||
// teamsRepository = new MongoTeamsRepository();
|
||||
// maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository();
|
||||
// } else {
|
||||
// throw new AppError({ message: "Unsupported database type", status: 500 });
|
||||
// }
|
||||
|
||||
const monitorsRepository = new MongoMonitorsRepository();
|
||||
const checksRepository = new MongoChecksRepository(logger);
|
||||
const geoChecksRepository = new MongoGeoChecksRepository(logger);
|
||||
@@ -234,7 +176,6 @@ export const initializeServices = async ({
|
||||
const incidentsRepository = new MongoIncidentsRepository();
|
||||
const teamsRepository = new MongoTeamsRepository();
|
||||
const maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository();
|
||||
const queueWorkersRepository = new MongoQueueWorkersRepository();
|
||||
const jobsRepository = new MongoJobsRepository();
|
||||
|
||||
// Inject settings repository into settings service (now that DB is connected)
|
||||
@@ -349,22 +290,7 @@ export const initializeServices = async ({
|
||||
geoCheckPipeline
|
||||
);
|
||||
|
||||
if (queueMode === "worker" && envSettings.queueType !== "lessSimpleQueue") {
|
||||
throw new AppError({
|
||||
message: `Worker mode requires QUEUE_TYPE="lessSimpleQueue" got ${envSettings.queueType}`,
|
||||
status: 500,
|
||||
service: "config services",
|
||||
method: "initializeServices",
|
||||
});
|
||||
}
|
||||
|
||||
let worker: IWorker;
|
||||
switch (envSettings.queueType) {
|
||||
case "lessSimpleQueue":
|
||||
worker = await LessSimpleWorker.create(logger, workerHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode);
|
||||
break;
|
||||
case "dbQueue":
|
||||
worker = await DBQueueWorker.create(
|
||||
const worker = await DBQueueWorker.create(
|
||||
logger,
|
||||
jobsRepository,
|
||||
monitorsRepository,
|
||||
@@ -374,13 +300,9 @@ export const initializeServices = async ({
|
||||
checkEvaluator,
|
||||
geoCheckPipeline,
|
||||
reactorDispatcher,
|
||||
workerHelper
|
||||
workerHelper,
|
||||
envSettings.queueMode
|
||||
);
|
||||
break;
|
||||
|
||||
default:
|
||||
worker = await SuperSimpleQueue.create(logger, workerHelper, monitorsRepository);
|
||||
}
|
||||
|
||||
// Business services
|
||||
const userService = new UserService({
|
||||
|
||||
@@ -1,18 +0,0 @@
|
||||
import { Schema, model } from "mongoose";
|
||||
import { QueueModes } from "@/domain/app-settings/app-settings.type.js";
|
||||
import { QueueWorkerDocument } from "@/domain/queue-workers/queue-worker.type.js";
|
||||
|
||||
const QueueWorkerSchema = new Schema<QueueWorkerDocument>(
|
||||
{
|
||||
_id: { type: String, required: true },
|
||||
mode: { type: String, enum: QueueModes, required: true },
|
||||
lastSeenAt: { type: Date, default: Date.now, expires: 30 },
|
||||
},
|
||||
{ timestamps: true }
|
||||
);
|
||||
|
||||
const QueueWorkerModel = model<QueueWorkerDocument>("QueueWorker", QueueWorkerSchema);
|
||||
|
||||
export type { QueueWorkerDocument };
|
||||
export { QueueWorkerModel };
|
||||
export default QueueWorkerModel;
|
||||
@@ -1,8 +0,0 @@
|
||||
import type { QueueMode } from "@/domain/app-settings/app-settings.type.js";
|
||||
import type { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js";
|
||||
|
||||
export interface IQueueWorkersRepository {
|
||||
upsert(workerId: string, mode: QueueMode): Promise<void>;
|
||||
findRecent(maxAgeMs: number): Promise<QueueWorker[]>;
|
||||
deleteById(workerId: string): Promise<void>;
|
||||
}
|
||||
@@ -1,33 +0,0 @@
|
||||
import type { QueueMode } from "@/domain/app-settings/app-settings.type.js";
|
||||
import type { IQueueWorkersRepository } from "./queue-worker.repository.interface.js";
|
||||
import type { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js";
|
||||
import { QueueWorkerModel, type QueueWorkerDocument } from "@/domain/queue-workers/queue-worker.model.js";
|
||||
const SERVICE_NAME = "MongoQueueWorkersRepository";
|
||||
|
||||
class MongoQueueWorkersRepository implements IQueueWorkersRepository {
|
||||
static SERVICE_NAME = SERVICE_NAME;
|
||||
|
||||
protected toEntity = (doc: QueueWorkerDocument): QueueWorker => {
|
||||
return {
|
||||
workerId: doc._id,
|
||||
mode: doc.mode,
|
||||
lastSeenAt: doc.lastSeenAt.getTime(),
|
||||
};
|
||||
};
|
||||
|
||||
upsert = async (workerId: string, mode: QueueMode): Promise<void> => {
|
||||
await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true });
|
||||
};
|
||||
|
||||
findRecent = async (maxAgeMs: number): Promise<QueueWorker[]> => {
|
||||
const cutoff = new Date(Date.now() - maxAgeMs);
|
||||
const docs = await QueueWorkerModel.find({ lastSeenAt: { $gte: cutoff } });
|
||||
return docs.map(this.toEntity);
|
||||
};
|
||||
|
||||
deleteById = async (workerId: string): Promise<void> => {
|
||||
await QueueWorkerModel.deleteOne({ _id: workerId });
|
||||
};
|
||||
}
|
||||
|
||||
export default MongoQueueWorkersRepository;
|
||||
@@ -1,18 +0,0 @@
|
||||
import { type QueueMode } from "@/domain/app-settings/app-settings.type.js";
|
||||
|
||||
// Heartbeat presence record for a scheduler instance. _id is the scheduler's
|
||||
// workerId (hostname:pid:uuid). The TTL index on lastSeenAt garbage-collects
|
||||
// records for instances that have stopped heartbeating.
|
||||
export interface QueueWorkerDocument {
|
||||
_id: string;
|
||||
mode: QueueMode;
|
||||
lastSeenAt: Date;
|
||||
createdAt: Date;
|
||||
updatedAt: Date;
|
||||
}
|
||||
|
||||
export type QueueWorker = {
|
||||
workerId: string; // hostname:pid:uuid
|
||||
mode: QueueMode;
|
||||
lastSeenAt: number; // epoch ms of the last heartbeat
|
||||
};
|
||||
+1
-1
@@ -24,7 +24,7 @@ const startApp = async () => {
|
||||
logger = new Logger({ envSettings });
|
||||
|
||||
// Initialize services (connects DB, creates repositories, injects settingsRepository)
|
||||
const services = await initializeServices({ logger, envSettings, settingsService, queueMode: envSettings.queueMode });
|
||||
const services = await initializeServices({ logger, envSettings, settingsService });
|
||||
|
||||
// If this is a worker instance, we're done. No need for express
|
||||
if (envSettings.queueMode === "worker") {
|
||||
|
||||
@@ -13,6 +13,7 @@ import { ICheckEvaluator } from "@/worker/worker.check-evaluator.js";
|
||||
import { ICheckPipeline } from "@/worker/worker.check-pipeline.js";
|
||||
import { IReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js";
|
||||
import { IWorkerHelper } from "@/worker/worker.helper.js";
|
||||
import { QueueMode, QueueModes } from "@/domain/app-settings/app-settings.type.js";
|
||||
const SERVICE_NAME = "JobQueue";
|
||||
const POLL_MS = 250;
|
||||
const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000;
|
||||
@@ -47,7 +48,8 @@ export class DBQueueWorker implements IWorker {
|
||||
private checkEvaluator: ICheckEvaluator,
|
||||
private geoCheckPipeline: ICheckPipeline,
|
||||
private dispatcher: IReactorDispatcher,
|
||||
private helper: IWorkerHelper
|
||||
private helper: IWorkerHelper,
|
||||
private mode: QueueMode
|
||||
) {}
|
||||
|
||||
get serviceName() {
|
||||
@@ -64,7 +66,8 @@ export class DBQueueWorker implements IWorker {
|
||||
checkEvaluator: ICheckEvaluator,
|
||||
geoCheckPipeline: ICheckPipeline,
|
||||
dispatcher: IReactorDispatcher,
|
||||
helper: IWorkerHelper
|
||||
helper: IWorkerHelper,
|
||||
mode: QueueMode
|
||||
): Promise<DBQueueWorker> {
|
||||
const instance = new DBQueueWorker(
|
||||
logger,
|
||||
@@ -76,7 +79,8 @@ export class DBQueueWorker implements IWorker {
|
||||
checkEvaluator,
|
||||
geoCheckPipeline,
|
||||
dispatcher,
|
||||
helper
|
||||
helper,
|
||||
mode
|
||||
);
|
||||
await instance.init();
|
||||
return instance;
|
||||
@@ -242,10 +246,14 @@ export class DBQueueWorker implements IWorker {
|
||||
|
||||
init = async () => {
|
||||
this.stopped = false;
|
||||
if (this.mode === "primary") {
|
||||
await this.reconcile();
|
||||
return true;
|
||||
} else {
|
||||
for (const type of Object.keys(this.inFlight) as JobType[]) {
|
||||
this.startLoop(type);
|
||||
}
|
||||
}
|
||||
return true;
|
||||
};
|
||||
addJob = async (_monitorId: string, monitor: Monitor) => {
|
||||
@@ -307,7 +315,7 @@ export class DBQueueWorker implements IWorker {
|
||||
jobsWithFailures: [],
|
||||
totalRuns: 0,
|
||||
totalFailures: 0,
|
||||
workers: [{ workerId: this.workerId, mode: "primary", lastSeenAt: now }],
|
||||
workers: [{ workerId: this.workerId, mode: this.mode, lastSeenAt: now }],
|
||||
}
|
||||
);
|
||||
};
|
||||
|
||||
@@ -1,427 +0,0 @@
|
||||
import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
|
||||
import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js";
|
||||
import { ILogger } from "@/utils/logger.js";
|
||||
import { MongoStore, Scheduler, IJob, AddJobInput } from "less-simple-scheduler";
|
||||
import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js";
|
||||
import { IWorkerHelper } from "@/worker/worker.helper.js";
|
||||
import { type WorkerMetrics, type WorkerJobsPagination, type WorkerJobsPage, IWorker } from "@/worker/worker.interface.js";
|
||||
import { type QueueMode } from "@/domain/app-settings/app-settings.type.js";
|
||||
import { EnvConfig } from "@/domain/app-settings/app-settings.service.js";
|
||||
const SERVICE_NAME = "JobQueue";
|
||||
|
||||
// How long after its last heartbeat a worker is still considered alive. Must be longer than the scheduler's heartbeat
|
||||
const WORKER_TTL_MS = 30_000;
|
||||
|
||||
// For discriminating job types
|
||||
type MonitorJob = Omit<IJob, "data"> & { data: Monitor };
|
||||
const MONITOR_TEMPLATES = new Set<string>(["monitor-job", "geo-check-job"]);
|
||||
const isMonitorJob = (job: IJob): job is MonitorJob => MONITOR_TEMPLATES.has(job.template);
|
||||
|
||||
export class LessSimpleWorker implements IWorker {
|
||||
static SERVICE_NAME = SERVICE_NAME;
|
||||
|
||||
private logger: ILogger;
|
||||
private helper: IWorkerHelper;
|
||||
private monitorsRepository: IMonitorsRepository;
|
||||
private workersRepository: IQueueWorkersRepository;
|
||||
private readonly scheduler: Scheduler;
|
||||
private queueMode: QueueMode;
|
||||
private heartbeatListener: ((workerId: string) => void) | null = null;
|
||||
|
||||
constructor(
|
||||
logger: ILogger,
|
||||
helper: IWorkerHelper,
|
||||
monitorsRepository: IMonitorsRepository,
|
||||
workersRepository: IQueueWorkersRepository,
|
||||
scheduler: Scheduler,
|
||||
queueMode: QueueMode
|
||||
) {
|
||||
this.logger = logger;
|
||||
this.helper = helper;
|
||||
this.monitorsRepository = monitorsRepository;
|
||||
this.workersRepository = workersRepository;
|
||||
this.scheduler = scheduler;
|
||||
this.queueMode = queueMode;
|
||||
this.registerListeners();
|
||||
}
|
||||
|
||||
get serviceName() {
|
||||
return LessSimpleWorker.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: `${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: `${workerId} ${job.id} locked`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:abort", (workerId, job, reason) => {
|
||||
this.logger.warn({
|
||||
message: `${workerId} ${job.id} aborted: ${reason}`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:attempt", (workerId, job, attempt) => {
|
||||
this.logger.debug({
|
||||
message: `${workerId} ${job.id} attempt ${attempt}`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:complete", (workerId, job) => {
|
||||
this.logger.debug({
|
||||
message: `${workerId} ${job.id} completed successfully`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:exhausted", (workerId, job, error) => {
|
||||
this.logger.error({
|
||||
message: `${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: `${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: `${workerId} ${job.id} started`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
static async create(
|
||||
logger: ILogger,
|
||||
helper: IWorkerHelper,
|
||||
monitorsRepository: IMonitorsRepository,
|
||||
workersRepository: IQueueWorkersRepository,
|
||||
envSettings: EnvConfig,
|
||||
queueMode: QueueMode
|
||||
) {
|
||||
const store = new MongoStore({
|
||||
url: envSettings.dbConnectionString ?? "mongodb://localhost:27017/uptime_db",
|
||||
});
|
||||
const scheduler = new Scheduler(store, {
|
||||
concurrency: envSettings.queueMode === "primary" ? 0 : 50,
|
||||
processEvery: 100,
|
||||
});
|
||||
const instance = new LessSimpleWorker(logger, helper, monitorsRepository, workersRepository, scheduler, queueMode);
|
||||
await instance.init(queueMode);
|
||||
|
||||
return instance;
|
||||
}
|
||||
|
||||
private startHeartbeat = (queueMode: QueueMode) => {
|
||||
const beat = (
|
||||
workerId: string // This is the heartbeat function that registers workers
|
||||
) =>
|
||||
this.workersRepository.upsert(workerId, queueMode).catch((error: unknown) => {
|
||||
this.logger.warn({
|
||||
message: `Worker heartbeat failed: ${error instanceof Error ? error.message : String(error)}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "startHeartbeat",
|
||||
});
|
||||
});
|
||||
beat(this.scheduler.workerId); // Beat immediately to register
|
||||
if (this.heartbeatListener) {
|
||||
this.scheduler.off("scheduler:heartbeat", this.heartbeatListener); // remove old listeners
|
||||
}
|
||||
this.heartbeatListener = beat; /// Keep a reference to the listener
|
||||
this.scheduler.on("scheduler:heartbeat", beat); // beat on every heartbeat
|
||||
};
|
||||
|
||||
init = async (queueMode: QueueMode = "primary") => {
|
||||
try {
|
||||
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());
|
||||
|
||||
// Register in the worker registry
|
||||
this.startHeartbeat(queueMode);
|
||||
|
||||
// If this is a worker, return early - primary will populate queue
|
||||
if (queueMode === "worker") {
|
||||
this.logger.info({
|
||||
message: `Worker queue: ${this.scheduler.workerId} initialized`,
|
||||
service: SERVICE_NAME,
|
||||
method: "init",
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
const monitors = await this.monitorsRepository.findAll();
|
||||
if (!monitors) {
|
||||
return true;
|
||||
}
|
||||
const jobInputs = monitors.flatMap((monitor) => this.buildJobInputs(monitor));
|
||||
if (jobInputs.length > 0) {
|
||||
await this.scheduler.addJobs(jobInputs);
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
};
|
||||
|
||||
// Builds the scheduler inputs for a monitor: always a monitor-job, plus a
|
||||
// geo-check job when the monitor supports and enables geo checks.
|
||||
private buildJobInputs = (monitor: Monitor): AddJobInput[] => {
|
||||
const inputs: AddJobInput[] = [
|
||||
{
|
||||
id: monitor.id,
|
||||
template: "monitor-job",
|
||||
repeat: monitor.interval,
|
||||
active: monitor.isActive,
|
||||
data: monitor,
|
||||
upsert: true,
|
||||
jitter: true,
|
||||
},
|
||||
];
|
||||
|
||||
if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) {
|
||||
inputs.push({
|
||||
id: `${monitor.id}-geo`,
|
||||
template: "geo-check-job",
|
||||
repeat: monitor.geoCheckInterval,
|
||||
active: monitor.isActive,
|
||||
data: monitor,
|
||||
upsert: true,
|
||||
jitter: true,
|
||||
});
|
||||
}
|
||||
|
||||
return inputs;
|
||||
};
|
||||
|
||||
addJob = async (_monitorId: string, monitor: Monitor) => {
|
||||
await this.scheduler.addJobs(this.buildJobInputs(monitor));
|
||||
};
|
||||
|
||||
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 () => {
|
||||
// Stop beating before we remove this worker, so a heartbeat firing between
|
||||
// the remove and the scheduler stopping can't re-insert it.
|
||||
if (this.heartbeatListener) {
|
||||
this.scheduler.off("scheduler:heartbeat", this.heartbeatListener);
|
||||
this.heartbeatListener = null;
|
||||
}
|
||||
await this.workersRepository.deleteById(this.scheduler.workerId).catch((error: unknown) => {
|
||||
this.logger.warn({
|
||||
message: `Failed to deregister worker on shutdown: ${error instanceof Error ? error.message : String(error)}`,
|
||||
service: SERVICE_NAME,
|
||||
method: "shutdown",
|
||||
});
|
||||
});
|
||||
this.scheduler.stop();
|
||||
};
|
||||
|
||||
getMetrics = async (): Promise<WorkerMetrics> => {
|
||||
const stats = await this.scheduler.getStats();
|
||||
|
||||
const jobsWithFailures = stats.jobsWithFailures.map((failure) => {
|
||||
// `data` is untyped at the store level: for monitor jobs it's the Monitor,
|
||||
// for system jobs (e.g. cleanup) it has no url/type => null
|
||||
const monitor = failure.data as Monitor | undefined;
|
||||
return {
|
||||
monitorId: failure.id,
|
||||
monitorUrl: monitor?.url ?? null,
|
||||
monitorType: monitor?.type ?? null,
|
||||
failedAt: failure.failedAt,
|
||||
failCount: failure.failCount,
|
||||
failReason: failure.failReason,
|
||||
};
|
||||
});
|
||||
|
||||
const workers = await this.workersRepository.findRecent(WORKER_TTL_MS);
|
||||
return {
|
||||
jobs: stats.jobs,
|
||||
activeJobs: stats.activeJobs,
|
||||
failingJobs: stats.failingJobs,
|
||||
totalRuns: stats.totalRuns,
|
||||
totalFailures: stats.totalFailures,
|
||||
jobsWithFailures,
|
||||
workers,
|
||||
};
|
||||
};
|
||||
|
||||
getJobs = async ({ page = 0, rowsPerPage = 0 }: WorkerJobsPagination): Promise<WorkerJobsPage> => {
|
||||
const listOptions = rowsPerPage > 0 ? { skip: Math.max(page, 0) * rowsPerPage, limit: rowsPerPage } : undefined;
|
||||
const [jobs, count] = await Promise.all([this.scheduler.getJobs(listOptions), this.scheduler.countJobs()]);
|
||||
const mapped = 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,
|
||||
lockedBy: job.lockedBy ?? null,
|
||||
lockedUntil: job.lockedUntil ?? 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,
|
||||
};
|
||||
});
|
||||
|
||||
return { jobs: mapped, count };
|
||||
};
|
||||
|
||||
flushQueues = async () => {
|
||||
if (this.queueMode !== "primary") {
|
||||
// Don't allow workers to flush, though this isn't currently reachable since workers have no API endpoints
|
||||
this.logger.warn({ message: "Ignoring flush on non-primary worker", service: SERVICE_NAME, method: "flushQueues" });
|
||||
return { success: false };
|
||||
}
|
||||
|
||||
const flushRes = await this.scheduler.flushJobs(); // Less Simple Queue must flush before stopping as it needs DB
|
||||
const stopRes = await this.scheduler.stop(); // Destructive, other workers will abort their jobs
|
||||
const initRes = await this.init();
|
||||
return {
|
||||
success: Boolean(stopRes && flushRes && initRes),
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -1,384 +0,0 @@
|
||||
import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
|
||||
import { ILogger } from "@/utils/logger.js";
|
||||
import Scheduler from "super-simple-scheduler";
|
||||
import { Monitor, MonitorType, supportsGeoCheck } from "@/domain/monitors/monitor.types.js";
|
||||
import { hostname } from "node:os";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { QueueMode } from "@/domain/app-settings/app-settings.type.js";
|
||||
import { IWorker, WorkerJobSummary } from "@/worker/worker.interface.js";
|
||||
import { IWorkerHelper } from "@/worker/worker.helper.js";
|
||||
|
||||
const SERVICE_NAME = "JobQueue";
|
||||
|
||||
export class SuperSimpleQueue implements IWorker {
|
||||
static SERVICE_NAME = SERVICE_NAME;
|
||||
|
||||
private logger: ILogger;
|
||||
private helper: IWorkerHelper;
|
||||
private workerId: string;
|
||||
private monitorsRepository: IMonitorsRepository;
|
||||
private readonly scheduler: Scheduler;
|
||||
|
||||
constructor(logger: ILogger, helper: IWorkerHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) {
|
||||
this.logger = logger;
|
||||
this.helper = helper;
|
||||
this.monitorsRepository = monitorsRepository;
|
||||
this.scheduler = scheduler;
|
||||
this.registerListeners();
|
||||
this.workerId = `${hostname()}:${process.pid}:${randomUUID()}`;
|
||||
}
|
||||
|
||||
get serviceName() {
|
||||
return SuperSimpleQueue.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", (error) => {
|
||||
this.logger.error({
|
||||
message: `Scheduler error: ${error instanceof Error ? error.message : String(error)}`,
|
||||
service: SERVICE_NAME,
|
||||
stack: error instanceof Error ? error.stack : undefined,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:abort", (job, reason) => {
|
||||
this.logger.warn({
|
||||
message: `${job.id} aborted: ${reason}`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:attempt", (job, attempt) => {
|
||||
this.logger.debug({
|
||||
message: `${job.id} attempt ${attempt}`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:complete", (job) => {
|
||||
this.logger.debug({
|
||||
message: `${job.id} completed successfully`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
|
||||
this.scheduler.on("job:exhausted", (job, error) => {
|
||||
this.logger.error({
|
||||
message: `${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", (job, error, attempt) => {
|
||||
this.logger.warn({
|
||||
message: `${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", (job) => {
|
||||
this.logger.debug({
|
||||
message: `${job.id} started`,
|
||||
service: SERVICE_NAME,
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
static async create(logger: ILogger, helper: IWorkerHelper, monitorsRepository: IMonitorsRepository) {
|
||||
const scheduler = new Scheduler({
|
||||
concurrency: 50,
|
||||
// storeType: "mongo",
|
||||
// storeType: "redis",
|
||||
// dbUri: envSettings.dbConnectionString,
|
||||
});
|
||||
const instance = new SuperSimpleQueue(logger, helper, monitorsRepository, scheduler);
|
||||
await instance.init();
|
||||
|
||||
return instance;
|
||||
}
|
||||
|
||||
init = async () => {
|
||||
try {
|
||||
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 });
|
||||
this.scheduler.addJob({ id: "cleanup-retention", template: "cleanup-retention-job", active: true, repeat: 24 * 60 * 60 * 1000 });
|
||||
|
||||
return true;
|
||||
} catch (error: unknown) {
|
||||
this.logger.error({
|
||||
message: `Failed to initialize SuperSimpleQueue: ${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,
|
||||
});
|
||||
|
||||
// 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,
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
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,
|
||||
});
|
||||
};
|
||||
|
||||
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: number;
|
||||
activeJobs: number;
|
||||
failingJobs: number;
|
||||
jobsWithFailures: Array<{
|
||||
monitorId: string | number;
|
||||
monitorUrl: string;
|
||||
monitorType: MonitorType;
|
||||
failedAt: number | null;
|
||||
failCount: number;
|
||||
failReason: string | null;
|
||||
}>;
|
||||
totalRuns: number;
|
||||
totalFailures: number;
|
||||
workers: Array<{
|
||||
workerId: string;
|
||||
mode: QueueMode;
|
||||
lastSeenAt: number;
|
||||
}>;
|
||||
} = jobs.reduce(
|
||||
(acc, job) => {
|
||||
const runCount = job.runCount ?? 0;
|
||||
const failCount = job.failCount ?? 0;
|
||||
const lastFailedAt = job.lastFailedAt ?? 0;
|
||||
const lastRunAt = job.lastRunAt ?? 0;
|
||||
acc.totalRuns += runCount;
|
||||
acc.totalFailures += failCount;
|
||||
acc.jobs++;
|
||||
if (failCount > 0 && lastFailedAt >= lastRunAt) {
|
||||
acc.failingJobs++;
|
||||
}
|
||||
|
||||
if (job.lockedAt) {
|
||||
acc.activeJobs++;
|
||||
}
|
||||
|
||||
if (failCount > 0) {
|
||||
acc.jobsWithFailures.push({
|
||||
monitorId: job.id,
|
||||
monitorUrl: job.data?.url || null,
|
||||
monitorType: job.data?.type || null,
|
||||
failedAt: job.lastFailedAt ?? null,
|
||||
failCount,
|
||||
failReason: job.lastFailReason ?? null,
|
||||
});
|
||||
}
|
||||
return acc;
|
||||
},
|
||||
{
|
||||
jobs: 0,
|
||||
activeJobs: 0,
|
||||
failingJobs: 0,
|
||||
jobsWithFailures: [] as Array<{
|
||||
monitorId: string | number;
|
||||
monitorUrl: string;
|
||||
monitorType: MonitorType;
|
||||
failedAt: number | null;
|
||||
failCount: number;
|
||||
failReason: string | null;
|
||||
}>,
|
||||
totalRuns: 0,
|
||||
totalFailures: 0,
|
||||
workers: [
|
||||
{
|
||||
workerId: this.workerId,
|
||||
mode: "primary",
|
||||
lastSeenAt: Date.now(),
|
||||
},
|
||||
],
|
||||
}
|
||||
);
|
||||
return metrics;
|
||||
};
|
||||
|
||||
getJobs = async ({ page = 0, rowsPerPage = 0 }: { page?: number; rowsPerPage?: number }): Promise<{ jobs: WorkerJobSummary[]; count: number }> => {
|
||||
const jobs = await this.scheduler.getJobs();
|
||||
const mapped = jobs.map((job) => {
|
||||
return {
|
||||
monitorId: job.id,
|
||||
monitorUrl: job.data?.url || null,
|
||||
monitorType: job.data?.type || null,
|
||||
monitorInterval: job.data?.interval || null,
|
||||
monitorGeoInterval: job.data?.geoCheckInterval || null,
|
||||
monitorActive: job.data?.isActive ?? null,
|
||||
active: job.active,
|
||||
repeat: job.repeat ?? null,
|
||||
lockedBy: job.lockedAt ? this.workerId : null,
|
||||
lockedUntil: null,
|
||||
lockedAt: job.lockedAt ?? null,
|
||||
runCount: job.runCount ?? 0,
|
||||
failCount: job.failCount ?? 0,
|
||||
failReason: job.lastFailReason ?? null,
|
||||
lastRunAt: job.lastRunAt ?? null,
|
||||
lastFinishedAt: job.lastFinishedAt ?? null,
|
||||
lastRunTook: job.lockedAt ? null : (job.lastFinishedAt ?? 0) - (job.lastRunAt ?? 0),
|
||||
lastFailedAt: job.lastFailedAt ?? null,
|
||||
};
|
||||
});
|
||||
|
||||
const start = Math.max(page, 0) * rowsPerPage;
|
||||
return {
|
||||
jobs: rowsPerPage > 0 ? mapped.slice(start, start + rowsPerPage) : mapped,
|
||||
count: mapped.length,
|
||||
};
|
||||
};
|
||||
|
||||
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),
|
||||
};
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user