diff --git a/server/src/api/controllers/queueController.ts b/server/src/api/controllers/queueController.ts index 0561ba6f0..0d3e2d1d0 100755 --- a/server/src/api/controllers/queueController.ts +++ b/server/src/api/controllers/queueController.ts @@ -1,4 +1,4 @@ -import { IJobQueue } from "@/service/job-queues/job-queue.interface.js"; +import { IWorker } from "@/worker/worker.interface.js"; import { getQueueJobsQueryValidation } from "@/api/validation/queueValidation.js"; import { Request, Response, NextFunction } from "express"; @@ -13,9 +13,9 @@ export interface IJobQueueController { class JobQueueController implements IJobQueueController { static SERVICE_NAME = SERVICE_NAME; - private jobQueue: IJobQueue; - constructor(jobQueue: IJobQueue) { - this.jobQueue = jobQueue; + private worker: IWorker; + constructor(worker: IWorker) { + this.worker = worker; } get serviceName() { @@ -24,7 +24,7 @@ class JobQueueController implements IJobQueueController { getMetrics = async (req: Request, res: Response, next: NextFunction) => { try { - const metrics = await this.jobQueue.getMetrics(); + const metrics = await this.worker.getMetrics(); res.status(200).json({ success: true, msg: "Queue metrics fetched successfully", @@ -38,7 +38,7 @@ class JobQueueController implements IJobQueueController { getJobs = async (req: Request, res: Response, next: NextFunction) => { try { const pagination = getQueueJobsQueryValidation.parse(req.query); - const { jobs, count } = await this.jobQueue.getJobs(pagination); + const { jobs, count } = await this.worker.getJobs(pagination); return res.status(200).json({ success: true, msg: "Queue jobs fetched successfully", @@ -52,8 +52,8 @@ class JobQueueController implements IJobQueueController { getAllMetrics = async (req: Request, res: Response, next: NextFunction) => { try { const pagination = getQueueJobsQueryValidation.parse(req.query); - const { jobs, count } = await this.jobQueue.getJobs(pagination); - const metrics = await this.jobQueue.getMetrics(); + const { jobs, count } = await this.worker.getJobs(pagination); + const metrics = await this.worker.getMetrics(); return res.status(200).json({ success: true, msg: "Queue metrics fetched successfully", @@ -66,7 +66,7 @@ class JobQueueController implements IJobQueueController { flushQueue = async (req: Request, res: Response, next: NextFunction) => { try { - const result = await this.jobQueue.flushQueues(); + const result = await this.worker.flushQueues(); return res.status(200).json({ success: true, msg: "Queue flushed successfully", diff --git a/server/src/config/controllers.ts b/server/src/config/controllers.ts index 09dc6651b..cdbb876b8 100644 --- a/server/src/config/controllers.ts +++ b/server/src/config/controllers.ts @@ -39,7 +39,7 @@ export const initializeControllers = (services: InitializedServices): Initialize geoCheckController: new GeoCheckController(services.geoChecksService), inviteController: new InviteController(services.inviteService), maintenanceWindowController: new MaintenanceWindowController(services.maintenanceWindowService), - queueController: new QueueController(services.jobQueue), + queueController: new QueueController(services.worker), logController: new LogController(services.logger), statusPageController: new StatusPageController(services.statusPageService, services.monitorsRepository, services.settingsService), notificationController: new NotificationController(services.notificationsService, services.monitorsRepository), diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 9aed9c968..dcd6cbb98 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -28,10 +28,10 @@ import { IUserService, UserService } from "@/domain/users/user.service.js"; import { BufferService, IBufferService } from "@/service/bufferService.js"; import { EmailService, IEmailService } from "@/service/emailService.js"; import { GlobalPingService } from "@/service/globalPingService.js"; -import { QueueHelper } from "@/service/job-queues/job-queue.helper.js"; -import { IJobQueue } from "@/service/job-queues/job-queue.interface.js"; -import { LessSimpleQueue } from "@/service/job-queues/job-queue.less-simple.js"; -import { SuperSimpleQueue } from "@/service/job-queues/job-queue.sumper-simple.js"; +import { QueueHelper } 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"; @@ -110,7 +110,7 @@ export type InitializedServices = { emailService: IEmailService; bufferService: IBufferService; statusService: IStatusService; - jobQueue: IJobQueue; + worker: IWorker; userService: IUserService; checkService: ICheckService; geoChecksService: IGeoChecksService; @@ -337,13 +337,13 @@ export const initializeServices = async ({ }); } - let jobQueue: IJobQueue; + let worker: IWorker; switch (envSettings.queueType) { case "lessSimpleQueue": - jobQueue = await LessSimpleQueue.create(logger, queueHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode); + worker = await LessSimpleWorker.create(logger, queueHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode); break; default: - jobQueue = await SuperSimpleQueue.create(logger, queueHelper, monitorsRepository); + worker = await SuperSimpleQueue.create(logger, queueHelper, monitorsRepository); } // Business services @@ -353,7 +353,7 @@ export const initializeServices = async ({ settingsService, logger, jwt, - jobQueue, + worker, monitorsRepository, usersRepository, invitesRepository, @@ -373,7 +373,7 @@ export const initializeServices = async ({ maintenanceWindowsRepository, }); const monitorService = new MonitorService({ - jobQueue, + worker, logger, games, monitorsRepository, @@ -393,7 +393,7 @@ export const initializeServices = async ({ emailService, bufferService, statusService, - jobQueue, + worker, userService, checkService, geoChecksService, diff --git a/server/src/domain/incidents/incident.service.ts b/server/src/domain/incidents/incident.service.ts index d588086b2..311fad567 100644 --- a/server/src/domain/incidents/incident.service.ts +++ b/server/src/domain/incidents/incident.service.ts @@ -8,7 +8,7 @@ import type { IMonitorsRepository } from "@/domain/monitors/monitor.repository.i import type { IUsersRepository } from "@/domain/users/user.repository.interface.js"; import type { Incident, IncidentSummary } from "@/domain/incidents/incident.type.js"; import type { User } from "@/domain/users/user.type.js"; -import type { MonitorActionDecision } from "@/service/job-queues/job-queue.helper.js"; +import type { MonitorActionDecision } from "@/worker/worker.helper.js"; import type { INotificationMessageBuilder } from "@/domain/notifications/notification.message-builder.js"; import type { ILogger } from "@/utils/logger.js"; diff --git a/server/src/domain/monitors/monitor.service.ts b/server/src/domain/monitors/monitor.service.ts index eaa40dcfe..a72d5a50d 100644 --- a/server/src/domain/monitors/monitor.service.ts +++ b/server/src/domain/monitors/monitor.service.ts @@ -21,7 +21,7 @@ import type { IStatusPagesRepository } from "@/domain/status-pages/status-page-r import demoMonitorsData from "@/utils/demoMonitors.json" with { type: "json" }; import { AppError } from "@/utils/AppError.js"; import type { ImportedMonitor } from "@/api/validation/monitorValidation.js"; -import { IJobQueue } from "@/service/job-queues/job-queue.interface.js"; +import { IWorker } from "@/worker/worker.interface.js"; import { ILogger } from "@/utils/logger.js"; const SERVICE_NAME = "MonitorService"; @@ -94,7 +94,7 @@ export interface IMonitorService { export class MonitorService implements IMonitorService { static SERVICE_NAME = SERVICE_NAME; - private jobQueue: IJobQueue; + private worker: IWorker; private logger: ILogger; private games: GamesMap; private monitorsRepository: IMonitorsRepository; @@ -105,7 +105,7 @@ export class MonitorService implements IMonitorService { private incidentsRepository: IIncidentsRepository; constructor({ - jobQueue, + worker, logger, games, monitorsRepository, @@ -115,7 +115,7 @@ export class MonitorService implements IMonitorService { statusPagesRepository, incidentsRepository, }: { - jobQueue: IJobQueue; + worker: IWorker; logger: ILogger; games: GamesMap; monitorsRepository: IMonitorsRepository; @@ -125,7 +125,7 @@ export class MonitorService implements IMonitorService { statusPagesRepository: IStatusPagesRepository; incidentsRepository: IIncidentsRepository; }) { - this.jobQueue = jobQueue; + this.worker = worker; this.logger = logger; this.games = games; this.monitorsRepository = monitorsRepository; @@ -171,7 +171,7 @@ export class MonitorService implements IMonitorService { throw new AppError({ message: "Failed to create monitor", status: 500, service: SERVICE_NAME, method: "createMonitor" }); } - this.jobQueue.addJob(monitor.id, monitor); + this.worker.addJob(monitor.id, monitor); }; createMonitors = async (monitors: Array): Promise => { @@ -180,7 +180,7 @@ export class MonitorService implements IMonitorService { throw new AppError({ message: "Failed to create monitors", status: 500, service: SERVICE_NAME, method: "createMonitors" }); } - await Promise.all(createdMonitors.map((monitor) => this.jobQueue.addJob(monitor.id, monitor))); + await Promise.all(createdMonitors.map((monitor) => this.worker.addJob(monitor.id, monitor))); return createdMonitors; }; @@ -196,7 +196,7 @@ export class MonitorService implements IMonitorService { })); const demoMonitors = await this.monitorsRepository.createMonitors(monitors as unknown as Monitor[]); - await Promise.all(demoMonitors.map((monitor) => this.jobQueue.addJob(monitor.id, monitor))); + await Promise.all(demoMonitors.map((monitor) => this.worker.addJob(monitor.id, monitor))); return demoMonitors; }; @@ -428,7 +428,7 @@ export class MonitorService implements IMonitorService { editMonitor = async ({ teamId, monitorId, body }: { teamId: string; monitorId: string; body: Partial }) => { const editedMonitor = await this.monitorsRepository.updateById(monitorId, teamId, body); - await this.jobQueue.updateJob(editedMonitor); + await this.worker.updateJob(editedMonitor); return editedMonitor; }; @@ -448,7 +448,7 @@ export class MonitorService implements IMonitorService { // If notifications were updated, we should update the jobs in the queue if (modifiedCount > 0) { const monitors = await this.monitorsRepository.findByIds(monitorIds); - await Promise.all(monitors.map((monitor) => this.jobQueue.updateJob(monitor))); + await Promise.all(monitors.map((monitor) => this.worker.updateJob(monitor))); } return modifiedCount; @@ -457,9 +457,9 @@ export class MonitorService implements IMonitorService { pauseMonitor = async ({ teamId, monitorId }: { teamId: string; monitorId: string }): Promise => { const monitor = await this.monitorsRepository.togglePauseById(monitorId, teamId); if (monitor.isActive) { - await this.jobQueue.resumeJob(monitor); + await this.worker.resumeJob(monitor); } else { - await this.jobQueue.pauseJob(monitor); + await this.worker.pauseJob(monitor); } return monitor; }; @@ -478,9 +478,9 @@ export class MonitorService implements IMonitorService { const results = await Promise.allSettled( monitors.map(async (monitor) => { if (monitor.isActive) { - await this.jobQueue.resumeJob(monitor); + await this.worker.resumeJob(monitor); } else { - await this.jobQueue.pauseJob(monitor); + await this.worker.pauseJob(monitor); } }) ); @@ -541,7 +541,7 @@ export class MonitorService implements IMonitorService { }); }); - await this.jobQueue.deleteJob(monitor); + await this.worker.deleteJob(monitor); return monitor; }; @@ -550,7 +550,7 @@ export class MonitorService implements IMonitorService { await Promise.all( monitors.map(async (monitor) => { try { - await this.jobQueue.deleteJob(monitor); + await this.worker.deleteJob(monitor); await this.checksRepository.deleteByMonitorId(monitor.id); await this.geoChecksRepository.deleteByMonitorId(monitor.id); await this.statusPagesRepository.removeMonitorFromStatusPages(monitor.id); diff --git a/server/src/domain/notifications/notification.message-builder.ts b/server/src/domain/notifications/notification.message-builder.ts index 9d3714ae7..b11d99a94 100644 --- a/server/src/domain/notifications/notification.message-builder.ts +++ b/server/src/domain/notifications/notification.message-builder.ts @@ -1,6 +1,6 @@ import type { Monitor } from "@/domain/monitors/monitor.types.js"; import type { HardwareStatusPayload, MonitorStatusResponse } from "@/types/network.js"; -import type { MonitorActionDecision } from "@/service/job-queues/job-queue.helper.js"; +import type { MonitorActionDecision } from "@/worker/worker.helper.js"; import type { NotificationMessage, NotificationType, diff --git a/server/src/domain/notifications/notification.service.ts b/server/src/domain/notifications/notification.service.ts index ea6a9265c..7fc6e3543 100644 --- a/server/src/domain/notifications/notification.service.ts +++ b/server/src/domain/notifications/notification.service.ts @@ -5,7 +5,7 @@ import type { NotificationMessage } from "@/domain/notifications/notification.ty import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; import { INotificationsRepository } from "@/domain/notifications/notification.repository.interface.js"; import { INotificationProvider } from "@/domain/notifications/providers/INotificationProvider.js"; -import type { MonitorActionDecision } from "@/service/job-queues/job-queue.helper.js"; +import type { MonitorActionDecision } from "@/worker/worker.helper.js"; import type { ISettingsService } from "@/domain/app-settings/app-settings.service.js"; import { ILogger } from "@/utils/logger.js"; import type { INotificationMessageBuilder } from "@/domain/notifications/notification.message-builder.js"; diff --git a/server/src/domain/users/user.service.ts b/server/src/domain/users/user.service.ts index 4edacfbc5..33d95f166 100644 --- a/server/src/domain/users/user.service.ts +++ b/server/src/domain/users/user.service.ts @@ -8,12 +8,12 @@ import type { User } from "@/domain/users/user.type.js"; import { canManageRole, type UserRole } from "@/domain/users/user.type.js"; import bcrypt from "bcryptjs"; import { AppError } from "@/utils/AppError.js"; -import { IJobQueue } from "@/service/job-queues/job-queue.interface.js"; import { IEmailService } from "@/service/emailService.js"; import { EnvConfig, ISettingsService } from "@/domain/app-settings/app-settings.service.js"; import { ILogger } from "@/utils/logger.js"; import jwt from "jsonwebtoken"; import crypto from "crypto"; +import { IWorker } from "@/worker/worker.interface.js"; type CryptoType = typeof crypto; type JwtType = typeof jwt; const SERVICE_NAME = "userService"; @@ -53,7 +53,7 @@ export class UserService implements IUserService { private settingsService: ISettingsService; private logger: ILogger; private jwt: JwtType; - private jobQueue: IJobQueue; + private worker: IWorker; private crypto: CryptoType; private monitorsRepository: IMonitorsRepository; private usersRepository: IUsersRepository; @@ -68,7 +68,7 @@ export class UserService implements IUserService { settingsService, logger, jwt, - jobQueue, + worker, monitorsRepository, usersRepository, invitesRepository, @@ -81,7 +81,7 @@ export class UserService implements IUserService { settingsService: ISettingsService; logger: ILogger; jwt: JwtType; - jobQueue: IJobQueue; + worker: IWorker; monitorsRepository: IMonitorsRepository; usersRepository: IUsersRepository; invitesRepository: IInvitesRepository; @@ -93,7 +93,7 @@ export class UserService implements IUserService { this.settingsService = settingsService; this.logger = logger; this.jwt = jwt; - this.jobQueue = jobQueue; + this.worker = worker; this.crypto = crypto; this.monitorsRepository = monitorsRepository; this.usersRepository = usersRepository; @@ -330,7 +330,7 @@ export class UserService implements IUserService { if (roles.includes("superadmin")) { const monitors = await this.monitorsRepository.findByTeamId(teamId, {}); if (monitors) { - await Promise.all(monitors.map((monitor) => this.jobQueue.deleteJob(monitor))); + await Promise.all(monitors.map((monitor) => this.worker.deleteJob(monitor))); } } // 6. Delete the user by id diff --git a/server/src/service/emailService.ts b/server/src/service/emailService.ts index ca153392d..c156cc18b 100755 --- a/server/src/service/emailService.ts +++ b/server/src/service/emailService.ts @@ -63,7 +63,7 @@ export class EmailService implements IEmailService { init = () => { const loadTemplate = (templateName: string) => { try { - const templatePath = this.path.join(__dirname, `../../templates/${templateName}.mjml`); + const templatePath = this.path.join(__dirname, `../templates/${templateName}.mjml`); const templateContent = this.fs.readFileSync(templatePath, "utf8"); return this.compile(templateContent); } catch (error: unknown) { diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index 3c1fc4ee7..38d1d5114 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -16,7 +16,7 @@ export const initShutdownListener = (server: Server | null, services: Initialize try { server?.close(); - await services.jobQueue.shutdown(); + await services.worker.shutdown(); await services.db.disconnect(); logger.info({ message: "Graceful shutdown complete" }); process.exit(0); diff --git a/server/src/service/job-queues/job-queue.helper.ts b/server/src/worker/worker.helper.ts similarity index 99% rename from server/src/service/job-queues/job-queue.helper.ts rename to server/src/worker/worker.helper.ts index 69ecb86c6..6840ff695 100644 --- a/server/src/service/job-queues/job-queue.helper.ts +++ b/server/src/worker/worker.helper.ts @@ -22,7 +22,7 @@ import { ITeamsRepository } from "@/domain/teams/team.repository.interface.js"; import { ILogger } from "@/utils/logger.js"; import { IBufferService } from "@/service/bufferService.js"; -export interface IQueueHelper { +export interface IWorkerHelper { readonly serviceName: string; getHeartbeatJob(): (monitor: Monitor) => Promise; getHeartbeatGeoJob(): (monitor: Monitor) => Promise; @@ -45,7 +45,7 @@ export interface MonitorActionDecision { }; } -export class QueueHelper implements IQueueHelper { +export class QueueHelper implements IWorkerHelper { static SERVICE_NAME = SERVICE_NAME; private logger: ILogger; diff --git a/server/src/service/job-queues/job-queue.interface.ts b/server/src/worker/worker.interface.ts similarity index 78% rename from server/src/service/job-queues/job-queue.interface.ts rename to server/src/worker/worker.interface.ts index 3e8267738..45dc94ca1 100644 --- a/server/src/service/job-queues/job-queue.interface.ts +++ b/server/src/worker/worker.interface.ts @@ -1,7 +1,7 @@ import { Monitor } from "@/domain/monitors/monitor.types.js"; import { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js"; -export type QueueJobFailure = { +export type WorkerJobFailure = { monitorId: string | number; monitorUrl: string | null; monitorType: string | null; @@ -10,17 +10,17 @@ export type QueueJobFailure = { failReason: string | null; }; -export type QueueMetrics = { +export type WorkerMetrics = { jobs: number; activeJobs: number; failingJobs: number; - jobsWithFailures: QueueJobFailure[]; + jobsWithFailures: WorkerJobFailure[]; totalRuns: number; totalFailures: number; workers: QueueWorker[]; }; -export type QueueJobSummary = { +export type WorkerJobSummary = { monitorId: string | number; monitorUrl: string | null; monitorType: string | null; @@ -41,17 +41,17 @@ export type QueueJobSummary = { repeat: number | null; }; -export type QueueJobsPagination = { +export type WorkerJobsPagination = { page?: number; rowsPerPage?: number; }; -export type QueueJobsPage = { - jobs: QueueJobSummary[]; +export type WorkerJobsPage = { + jobs: WorkerJobSummary[]; count: number; }; -export interface IJobQueue { +export interface IWorker { readonly serviceName: string; init(): Promise; addJob(monitorId: string, monitor: Monitor): Promise; @@ -60,7 +60,7 @@ export interface IJobQueue { resumeJob(monitor: Monitor): Promise; updateJob(monitor: Monitor): Promise; shutdown(): Promise; - getMetrics(): Promise; - getJobs(pagination: QueueJobsPagination): Promise; + getMetrics(): Promise; + getJobs(pagination: WorkerJobsPagination): Promise; flushQueues(): Promise<{ success: boolean }>; } diff --git a/server/src/service/job-queues/job-queue.less-simple.ts b/server/src/worker/worker.less-simple.ts similarity index 95% rename from server/src/service/job-queues/job-queue.less-simple.ts rename to server/src/worker/worker.less-simple.ts index ad12341c1..bd4752c6e 100644 --- a/server/src/service/job-queues/job-queue.less-simple.ts +++ b/server/src/worker/worker.less-simple.ts @@ -2,9 +2,9 @@ import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interf 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 { IQueueHelper } from "@/service/job-queues/job-queue.helper.js"; import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; -import { type QueueMetrics, type QueueJobsPagination, type QueueJobsPage, IJobQueue } from "@/service/job-queues/job-queue.interface.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"; @@ -17,11 +17,11 @@ 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 { +export class LessSimpleWorker implements IWorker { static SERVICE_NAME = SERVICE_NAME; private logger: ILogger; - private helper: IQueueHelper; + private helper: IWorkerHelper; private monitorsRepository: IMonitorsRepository; private workersRepository: IQueueWorkersRepository; private readonly scheduler: Scheduler; @@ -30,7 +30,7 @@ export class LessSimpleQueue implements IJobQueue { constructor( logger: ILogger, - helper: IQueueHelper, + helper: IWorkerHelper, monitorsRepository: IMonitorsRepository, workersRepository: IQueueWorkersRepository, scheduler: Scheduler, @@ -46,7 +46,7 @@ export class LessSimpleQueue implements IJobQueue { } get serviceName() { - return LessSimpleQueue.SERVICE_NAME; + return LessSimpleWorker.SERVICE_NAME; } private registerListeners = () => { @@ -126,7 +126,7 @@ export class LessSimpleQueue implements IJobQueue { static async create( logger: ILogger, - helper: IQueueHelper, + helper: IWorkerHelper, monitorsRepository: IMonitorsRepository, workersRepository: IQueueWorkersRepository, envSettings: EnvConfig, @@ -139,7 +139,7 @@ export class LessSimpleQueue implements IJobQueue { concurrency: envSettings.queueMode === "primary" ? 0 : 50, processEvery: 100, }); - const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler, queueMode); + const instance = new LessSimpleWorker(logger, helper, monitorsRepository, workersRepository, scheduler, queueMode); await instance.init(queueMode); return instance; @@ -350,7 +350,7 @@ export class LessSimpleQueue implements IJobQueue { this.scheduler.stop(); }; - getMetrics = async (): Promise => { + getMetrics = async (): Promise => { const stats = await this.scheduler.getStats(); const jobsWithFailures = stats.jobsWithFailures.map((failure) => { @@ -379,7 +379,7 @@ export class LessSimpleQueue implements IJobQueue { }; }; - getJobs = async ({ page = 0, rowsPerPage = 0 }: QueueJobsPagination): Promise => { + getJobs = async ({ page = 0, rowsPerPage = 0 }: WorkerJobsPagination): Promise => { 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) => { diff --git a/server/src/service/job-queues/job-queue.sumper-simple.ts b/server/src/worker/worker.super-simple.ts similarity index 95% rename from server/src/service/job-queues/job-queue.sumper-simple.ts rename to server/src/worker/worker.super-simple.ts index c53ca5860..86b9f152a 100644 --- a/server/src/service/job-queues/job-queue.sumper-simple.ts +++ b/server/src/worker/worker.super-simple.ts @@ -1,25 +1,25 @@ import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; import { ILogger } from "@/utils/logger.js"; import Scheduler from "super-simple-scheduler"; -import { IQueueHelper } from "@/service/job-queues/job-queue.helper.js"; 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 { IJobQueue, type QueueJobSummary } from "@/service/job-queues/job-queue.interface.js"; +import { IWorker, WorkerJobSummary } from "@/worker/worker.interface.js"; +import { IWorkerHelper } from "@/worker/worker.helper.js"; const SERVICE_NAME = "JobQueue"; -export class SuperSimpleQueue implements IJobQueue { +export class SuperSimpleQueue implements IWorker { static SERVICE_NAME = SERVICE_NAME; private logger: ILogger; - private helper: IQueueHelper; + private helper: IWorkerHelper; private workerId: string; private monitorsRepository: IMonitorsRepository; private readonly scheduler: Scheduler; - constructor(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) { + constructor(logger: ILogger, helper: IWorkerHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) { this.logger = logger; this.helper = helper; this.monitorsRepository = monitorsRepository; @@ -100,7 +100,7 @@ export class SuperSimpleQueue implements IJobQueue { }); }; - static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository) { + static async create(logger: ILogger, helper: IWorkerHelper, monitorsRepository: IMonitorsRepository) { const scheduler = new Scheduler({ concurrency: 50, // storeType: "mongo", @@ -341,7 +341,7 @@ export class SuperSimpleQueue implements IJobQueue { return metrics; }; - getJobs = async ({ page = 0, rowsPerPage = 0 }: { page?: number; rowsPerPage?: number }): Promise<{ jobs: QueueJobSummary[]; count: number }> => { + 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 {