From c9b8f434cd5c880c3351db76a16dd91a8750b8e7 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 23 Jun 2026 11:22:37 -0700 Subject: [PATCH 1/6] split worker into a scheduler and worker --- server/src/api/controllers/queueController.ts | 17 +- server/src/config/services.ts | 10 +- .../maintenance-window.service.ts | 14 +- server/src/domain/monitors/monitor.service.ts | 32 ++-- server/src/domain/users/user.service.ts | 12 +- server/src/worker/worker.db-queue.ts | 152 ++-------------- server/src/worker/worker.interface.ts | 10 +- server/src/worker/worker.job-scheduler.ts | 167 ++++++++++++++++++ 8 files changed, 227 insertions(+), 187 deletions(-) create mode 100644 server/src/worker/worker.job-scheduler.ts diff --git a/server/src/api/controllers/queueController.ts b/server/src/api/controllers/queueController.ts index 0d3e2d1d0..0a21761ab 100755 --- a/server/src/api/controllers/queueController.ts +++ b/server/src/api/controllers/queueController.ts @@ -1,4 +1,4 @@ -import { IWorker } from "@/worker/worker.interface.js"; +import { IJobScheduler } from "@/worker/worker.interface.js"; import { getQueueJobsQueryValidation } from "@/api/validation/queueValidation.js"; import { Request, Response, NextFunction } from "express"; @@ -13,10 +13,7 @@ export interface IJobQueueController { class JobQueueController implements IJobQueueController { static SERVICE_NAME = SERVICE_NAME; - private worker: IWorker; - constructor(worker: IWorker) { - this.worker = worker; - } + constructor(private scheduler: IJobScheduler) {} get serviceName() { return JobQueueController.SERVICE_NAME; @@ -24,7 +21,7 @@ class JobQueueController implements IJobQueueController { getMetrics = async (req: Request, res: Response, next: NextFunction) => { try { - const metrics = await this.worker.getMetrics(); + const metrics = await this.scheduler.getMetrics(); res.status(200).json({ success: true, msg: "Queue metrics fetched successfully", @@ -38,7 +35,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.worker.getJobs(pagination); + const { jobs, count } = await this.scheduler.getJobs(pagination); return res.status(200).json({ success: true, msg: "Queue jobs fetched successfully", @@ -52,8 +49,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.worker.getJobs(pagination); - const metrics = await this.worker.getMetrics(); + const { jobs, count } = await this.scheduler.getJobs(pagination); + const metrics = await this.scheduler.getMetrics(); return res.status(200).json({ success: true, msg: "Queue metrics fetched successfully", @@ -66,7 +63,7 @@ class JobQueueController implements IJobQueueController { flushQueue = async (req: Request, res: Response, next: NextFunction) => { try { - const result = await this.worker.flushQueues(); + const result = await this.scheduler.flushQueues(); return res.status(200).json({ success: true, msg: "Queue flushed successfully", diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 2f1826241..7bc59f881 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -31,7 +31,7 @@ import { BufferService, IBufferService } from "@/service/bufferService.js"; 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 { IQueueWorker } from "@/worker/worker.interface.js"; import { INetworkService, NetworkService } from "@/service/networkService.js"; import { IStatusService, StatusService } from "@/service/statusService.js"; import { MonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js"; @@ -117,7 +117,7 @@ export type InitializedServices = { emailService: IEmailService; bufferService: IBufferService; statusService: IStatusService; - worker: IWorker; + worker: IQueueWorker; userService: IUserService; checkService: ICheckService; geoChecksService: IGeoChecksService; @@ -315,7 +315,7 @@ export const initializeServices = async ({ settingsService, logger, jwt, - worker, + scheduler: worker, monitorsRepository, usersRepository, invitesRepository, @@ -334,10 +334,10 @@ export const initializeServices = async ({ monitorsRepository, maintenanceWindowsRepository, jobsRepository, - worker, + scheduler: worker, }); const monitorService = new MonitorService({ - worker, + scheduler: worker, logger, games, monitorsRepository, diff --git a/server/src/domain/maintenance-windows/maintenance-window.service.ts b/server/src/domain/maintenance-windows/maintenance-window.service.ts index b8601321f..5551aef62 100644 --- a/server/src/domain/maintenance-windows/maintenance-window.service.ts +++ b/server/src/domain/maintenance-windows/maintenance-window.service.ts @@ -1,10 +1,10 @@ import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js"; import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; -import { IWorker } from "@/worker/worker.interface.js"; import type { DurationUnit, MaintenanceWindow } from "@/domain/maintenance-windows/maintenance-window.type.js"; import { AppError } from "@/utils/AppError.js"; import { isWindowActive } from "@/utils/maintenanceWindow.js"; +import { IJobScheduler } from "@/worker/worker.interface.js"; const SERVICE_NAME = "maintenanceWindowService"; @@ -43,23 +43,23 @@ export class MaintenanceWindowService implements IMaintenanceWindowService { private monitorsRepository: IMonitorsRepository; private maintenanceWindowsRepository: IMaintenanceWindowsRepository; private jobsRepository: IJobsRepository; - private worker: IWorker; + private scheduler: IJobScheduler; constructor({ monitorsRepository, maintenanceWindowsRepository, jobsRepository, - worker, + scheduler, }: { monitorsRepository: IMonitorsRepository; maintenanceWindowsRepository: IMaintenanceWindowsRepository; jobsRepository: IJobsRepository; - worker: IWorker; + scheduler: IJobScheduler; }) { this.monitorsRepository = monitorsRepository; this.maintenanceWindowsRepository = maintenanceWindowsRepository; this.jobsRepository = jobsRepository; - this.worker = worker; + this.scheduler = scheduler; } get serviceName() { @@ -84,8 +84,8 @@ export class MaintenanceWindowService implements IMaintenanceWindowService { // Rearm jobs to run soon (now + jitter to avoid herding) await this.jobsRepository.markMonitorsDue(toInitializing, now.getTime()); // Wake the (possibly idle, backed-off) loops so the rearmed jobs run promptly instead of waiting up to POLL_MAX_MS - this.worker.wake("check"); - this.worker.wake("geo-check"); + this.scheduler.wake("check"); + this.scheduler.wake("geo-check"); } }; diff --git a/server/src/domain/monitors/monitor.service.ts b/server/src/domain/monitors/monitor.service.ts index 8fce77723..455a1daff 100644 --- a/server/src/domain/monitors/monitor.service.ts +++ b/server/src/domain/monitors/monitor.service.ts @@ -21,8 +21,8 @@ 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 { IWorker } from "@/worker/worker.interface.js"; import { ILogger } from "@/utils/logger.js"; +import { IJobScheduler } from "@/worker/worker.interface.js"; const SERVICE_NAME = "MonitorService"; type DateRangeKey = "recent" | "day" | "week" | "month" | "all"; @@ -94,7 +94,7 @@ export interface IMonitorService { export class MonitorService implements IMonitorService { static SERVICE_NAME = SERVICE_NAME; - private worker: IWorker; + private scheduler: IJobScheduler; private logger: ILogger; private games: GamesMap; private monitorsRepository: IMonitorsRepository; @@ -105,7 +105,7 @@ export class MonitorService implements IMonitorService { private incidentsRepository: IIncidentsRepository; constructor({ - worker, + scheduler, logger, games, monitorsRepository, @@ -115,7 +115,7 @@ export class MonitorService implements IMonitorService { statusPagesRepository, incidentsRepository, }: { - worker: IWorker; + scheduler: IJobScheduler; logger: ILogger; games: GamesMap; monitorsRepository: IMonitorsRepository; @@ -125,7 +125,7 @@ export class MonitorService implements IMonitorService { statusPagesRepository: IStatusPagesRepository; incidentsRepository: IIncidentsRepository; }) { - this.worker = worker; + this.scheduler = scheduler; 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.worker.addJob(monitor.id, monitor); + this.scheduler.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.worker.addJob(monitor.id, monitor))); + await Promise.all(createdMonitors.map((monitor) => this.scheduler.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.worker.addJob(monitor.id, monitor))); + await Promise.all(demoMonitors.map((monitor) => this.scheduler.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.worker.updateJob(editedMonitor); + await this.scheduler.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.worker.updateJob(monitor))); + await Promise.all(monitors.map((monitor) => this.scheduler.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.worker.resumeJob(monitor); + await this.scheduler.resumeJob(monitor); } else { - await this.worker.pauseJob(monitor); + await this.scheduler.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.worker.resumeJob(monitor); + await this.scheduler.resumeJob(monitor); } else { - await this.worker.pauseJob(monitor); + await this.scheduler.pauseJob(monitor); } }) ); @@ -541,7 +541,7 @@ export class MonitorService implements IMonitorService { }); }); - await this.worker.deleteJob(monitor); + await this.scheduler.deleteJob(monitor); return monitor; }; @@ -550,7 +550,7 @@ export class MonitorService implements IMonitorService { await Promise.all( monitors.map(async (monitor) => { try { - await this.worker.deleteJob(monitor); + await this.scheduler.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/users/user.service.ts b/server/src/domain/users/user.service.ts index 33d95f166..7bdb9677f 100644 --- a/server/src/domain/users/user.service.ts +++ b/server/src/domain/users/user.service.ts @@ -11,9 +11,9 @@ import { AppError } from "@/utils/AppError.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 { IJobScheduler } from "@/worker/worker.interface.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 worker: IWorker; + private scheduler: IJobScheduler; private crypto: CryptoType; private monitorsRepository: IMonitorsRepository; private usersRepository: IUsersRepository; @@ -68,7 +68,7 @@ export class UserService implements IUserService { settingsService, logger, jwt, - worker, + scheduler, monitorsRepository, usersRepository, invitesRepository, @@ -81,7 +81,7 @@ export class UserService implements IUserService { settingsService: ISettingsService; logger: ILogger; jwt: JwtType; - worker: IWorker; + scheduler: IJobScheduler; 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.worker = worker; + this.scheduler = scheduler; 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.worker.deleteJob(monitor))); + await Promise.all(monitors.map((monitor) => this.scheduler.deleteJob(monitor))); } } // 6. Delete the user by id diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 5e1a74bb4..ec7d1cf94 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -1,5 +1,5 @@ -import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; -import { IWorker, WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js"; +import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; +import { IQueueWorker } from "@/worker/worker.interface.js"; import { type Job, type JobSeed, type JobType, jobId, LOCK_MS } from "@/domain/jobs/job.type.js"; import { ILogger } from "@/utils/logger.js"; import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; @@ -14,6 +14,7 @@ import { IWorkerHelper } from "@/worker/worker.helper.js"; import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js"; import { WORKER_TTL_SECONDS } from "@/domain/queue-workers/queue-worker.model.js"; import { QueueMode } from "@/domain/app-settings/app-settings.type.js"; +import { JobScheduler } from "@/worker/worker.job-scheduler.js"; const SERVICE_NAME = "JobQueue"; const POLL_MS = 250; // base poll interval while a loop is actively claiming work const POLL_MAX_MS = 5000; // back off poll interval to 5s if no jobs @@ -29,10 +30,8 @@ const CONCURRENCY: Record = { "cleanup-retention": 1, }; -export class DBQueueWorker implements IWorker { +export class DBQueueWorker extends JobScheduler implements IQueueWorker { static SERVICE_NAME = SERVICE_NAME; - private stopped = false; - private timers = new Map(); private inFlight: Record = { check: 0, "geo-check": 0, @@ -41,33 +40,24 @@ export class DBQueueWorker implements IWorker { "cleanup-retention": 0, }; - // Per job type poll intervals. These back off on idle and reset on work or wake() - private pollMs: Record = { - check: POLL_MS, - "geo-check": POLL_MS, - evaluate: POLL_MS, - "cleanup-orphaned": POLL_MS, - "cleanup-retention": POLL_MS, - }; - private ticking: Partial> = {}; // Whether a loop is mid tick - private tickFns = new Map void>(); // Store tick fns so we can run them immediately on wake - constructor( private logger: ILogger, - private jobsRepository: IJobsRepository, + jobsRepository: IJobsRepository, // ← no modifier: forwarded to super private monitorsRepository: IMonitorsRepository, private checksRepository: IChecksRepository, - private checkService: ICheckService, // toStatusResponse + toLastEvaluatedAt + private checkService: ICheckService, private checkProducer: ICheckProducer, private checkEvaluator: ICheckEvaluator, private geoCheckPipeline: ICheckPipeline, private dispatcher: IReactorDispatcher, private helper: IWorkerHelper, - private queueWorkersRepository: IQueueWorkersRepository, + queueWorkersRepository: IQueueWorkersRepository, // ← no modifier: forwarded private queueMode: QueueMode, private queuePrimaryProcesses: boolean, - private readonly workerId: string // same id MongoJobsRepository stamps into lockedBy - ) {} + workerId: string + ) { + super(jobsRepository, queueWorkersRepository, workerId); + } get serviceName() { return DBQueueWorker.SERVICE_NAME; @@ -109,23 +99,6 @@ export class DBQueueWorker implements IWorker { return instance; } - // Helpers, builds job seeds. immediate=true runs on the next tick; otherwise jitter spreads the herd. - private toCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({ - id: jobId("check", monitor.id), - type: "check", - refId: monitor.id, - isActive: monitor.isActive, - nextScheduledAt: immediate ? now : now + Math.floor(Math.random() * monitor.interval), // Avoid herd - intervalMs: monitor.interval, - }); - - private toGeoCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({ - ...this.toCheckJob(monitor, now, immediate), - id: jobId("geo-check", monitor.id), - type: "geo-check", - intervalMs: monitor.geoCheckInterval!, - }); - private toCleanupJob = (type: "cleanup-orphaned" | "cleanup-retention", now: number): JobSeed => ({ id: jobId(type, null), type, @@ -135,20 +108,6 @@ export class DBQueueWorker implements IWorker { intervalMs: CLEANUP_INTERVAL_MS, }); - private toSummary = (job: Job): WorkerJobSummary => ({ - monitorId: job.refId ?? job.id, - monitorType: job.type, - monitorInterval: job.intervalMs, - monitorActive: job.isActive, - lockedBy: job.lockedBy, - lockedUntil: job.lockedUntil, - nextScheduledAt: job.nextScheduledAt, - runCount: job.runCount, - failCount: job.failCount, - failReason: job.lastFailReason, - lastFinishedAt: job.lastFinishedAt, - }); - // ******************** // Stage 1: This can eventually be offloaded to a separate service // ******************** @@ -276,15 +235,6 @@ export class DBQueueWorker implements IWorker { // Wake a worker loop after adding a job that's due immediately. // Otherwise the loop could be delayed by POLL_MAX_MS if it is already fully backed off - wake = (type: JobType) => { - this.pollMs[type] = POLL_MS; - if (this.stopped || this.ticking[type]) return; // If it's not running or mid tick, do nothing - const tick = this.tickFns.get(type); - if (!tick) return; // Try to get a handle on the tick fn, if we can't we're done - const pending = this.timers.get(type); - if (pending) clearTimeout(pending); // Clear out scheduled tick if any - this.timers.set(type, setTimeout(tick, 0)); // Set immediate tick - }; // Seed queue private reconcile = async () => { @@ -332,85 +282,7 @@ export class DBQueueWorker implements IWorker { return true; }; - addJob = async (_monitorId: string, monitor: Monitor) => { - const now = Date.now(); - await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now, true)); - this.wake("check"); - if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { - await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now, true)); - this.wake("geo-check"); - } - }; - deleteJob = async (monitor: Monitor) => { - await this.jobsRepository.deleteById(monitor.id); - }; - pauseJob = async (monitor: Monitor) => { - await this.jobsRepository.setActiveById(monitor.id, false); - }; - resumeJob = async (monitor: Monitor) => { - await this.jobsRepository.setActiveById(monitor.id, true); // a paused job's nextScheduledAt is now in the past, so it's due immediately - this.wake("check"); - if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) this.wake("geo-check"); - }; - updateJob = async (monitor: Monitor) => { - await this.jobsRepository.updateScheduleById(monitor.id, "check", monitor.interval); - if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { - await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, Date.now(), true)); // a newly enabled geo check should run right away - this.wake("geo-check"); - } else { - await this.jobsRepository.deleteByIdAndType(monitor.id, "geo-check"); - } - }; - shutdown = async () => { - this.stopped = true; - // Stop all timers - for (const timer of this.timers.values()) { - clearTimeout(timer); - } - this.timers.clear(); // Clear the map out - // Remove workers from registry - await this.queueWorkersRepository.deleteById(this.workerId); - }; - getMetrics = async (): Promise => { - const rows = await this.jobsRepository.findAll(); - const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); - const now = Date.now(); - return rows.reduce( - (acc, job) => { - acc.jobs++; - acc.totalRuns += job.runCount; - acc.totalFailures += job.failCount; - if (job.lockedBy !== null && job.lockedUntil !== null && job.lockedUntil > now) acc.activeJobs++; - if (job.failCount > 0) { - acc.failingJobs++; - acc.jobsWithFailures.push({ - monitorId: job.refId ?? job.id, - monitorUrl: null, - monitorType: job.type, - failedAt: job.lastFinishedAt, - failCount: job.failCount, - failReason: job.lastFailReason, - }); - } - return acc; - }, - { - jobs: 0, - activeJobs: 0, - failingJobs: 0, - jobsWithFailures: [], - totalRuns: 0, - totalFailures: 0, - workers, - } - ); - }; - - getJobs = async (pagination: WorkerJobsPagination) => { - const { jobs, count } = await this.jobsRepository.findPage(pagination); - return { jobs: jobs.map(this.toSummary), count }; - }; - flushQueues = async () => { + override flushQueues = async () => { await this.shutdown(); const ok = await this.init(); return { success: ok }; diff --git a/server/src/worker/worker.interface.ts b/server/src/worker/worker.interface.ts index 6dd360e50..522ed3204 100644 --- a/server/src/worker/worker.interface.ts +++ b/server/src/worker/worker.interface.ts @@ -56,17 +56,21 @@ export type WorkerJobsPage = { count: number; }; -export interface IWorker { +export interface IJobScheduler { 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; wake(type: JobType): void; - shutdown(): Promise; getMetrics(): Promise; getJobs(pagination: WorkerJobsPagination): Promise; flushQueues(): Promise<{ success: boolean }>; + shutdown(): Promise; +} + +export interface IQueueWorker extends IJobScheduler { + readonly serviceName: string; + init(): Promise; } diff --git a/server/src/worker/worker.job-scheduler.ts b/server/src/worker/worker.job-scheduler.ts new file mode 100644 index 000000000..4e05e4c38 --- /dev/null +++ b/server/src/worker/worker.job-scheduler.ts @@ -0,0 +1,167 @@ +import { IJobScheduler } from "@/worker/worker.interface.js"; +import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; +import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; +import { type Job, type JobSeed, type JobType, jobId } from "@/domain/jobs/job.type.js"; +import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js"; +import { WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js"; +import { WORKER_TTL_SECONDS } from "@/domain/queue-workers/queue-worker.model.js"; + +const POLL_MS = 250; // base poll interval while a loop is actively claiming work +const SERVICE_NAME = "JobScheduler"; +const WORKER_STALE_MS = WORKER_TTL_SECONDS * 1000; // a worker counts as alive if seen within this window + +export class JobScheduler implements IJobScheduler { + static SERVICE_NAME = SERVICE_NAME; + constructor( + protected jobsRepository: IJobsRepository, + protected queueWorkersRepository: IQueueWorkersRepository, + protected readonly workerId: string // same id MongoJobsRepository stamps into lockedBy + ) {} + + get serviceName() { + return JobScheduler.SERVICE_NAME; + } + + protected stopped = false; + protected ticking: Partial> = {}; // Whether a loop is mid tick + protected tickFns = new Map void>(); // Store tick fns so we can run them immediately on wake + protected timers = new Map(); + + // Per job type poll intervals. These back off on idle and reset on work or wake() + protected pollMs: Record = { + check: POLL_MS, + "geo-check": POLL_MS, + evaluate: POLL_MS, + "cleanup-orphaned": POLL_MS, + "cleanup-retention": POLL_MS, + }; + // Helpers, builds job seeds. immediate=true runs on the next tick; otherwise jitter spreads the herd. + protected toCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({ + id: jobId("check", monitor.id), + type: "check", + refId: monitor.id, + isActive: monitor.isActive, + nextScheduledAt: immediate ? now : now + Math.floor(Math.random() * monitor.interval), // Avoid herd + intervalMs: monitor.interval, + }); + + protected toGeoCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({ + ...this.toCheckJob(monitor, now, immediate), + id: jobId("geo-check", monitor.id), + type: "geo-check", + intervalMs: monitor.geoCheckInterval!, + }); + + private toSummary = (job: Job): WorkerJobSummary => ({ + monitorId: job.refId ?? job.id, + monitorType: job.type, + monitorInterval: job.intervalMs, + monitorActive: job.isActive, + lockedBy: job.lockedBy, + lockedUntil: job.lockedUntil, + nextScheduledAt: job.nextScheduledAt, + runCount: job.runCount, + failCount: job.failCount, + failReason: job.lastFailReason, + lastFinishedAt: job.lastFinishedAt, + }); + wake = (type: JobType) => { + this.pollMs[type] = POLL_MS; + if (this.stopped || this.ticking[type]) return; // If it's not running or mid tick, do nothing + const tick = this.tickFns.get(type); + if (!tick) return; // Try to get a handle on the tick fn, if we can't we're done + const pending = this.timers.get(type); + if (pending) clearTimeout(pending); // Clear out scheduled tick if any + this.timers.set(type, setTimeout(tick, 0)); // Set immediate tick + }; + + addJob = async (_monitorId: string, monitor: Monitor) => { + const now = Date.now(); + await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now, true)); + this.wake("check"); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { + await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now, true)); + this.wake("geo-check"); + } + }; + + deleteJob = async (monitor: Monitor) => { + await this.jobsRepository.deleteById(monitor.id); + }; + + pauseJob = async (monitor: Monitor) => { + await this.jobsRepository.setActiveById(monitor.id, false); + }; + + resumeJob = async (monitor: Monitor) => { + await this.jobsRepository.setActiveById(monitor.id, true); // a paused job's nextScheduledAt is now in the past, so it's due immediately + this.wake("check"); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) this.wake("geo-check"); + }; + + updateJob = async (monitor: Monitor) => { + await this.jobsRepository.updateScheduleById(monitor.id, "check", monitor.interval); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { + await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, Date.now(), true)); // a newly enabled geo check should run right away + this.wake("geo-check"); + } else { + await this.jobsRepository.deleteByIdAndType(monitor.id, "geo-check"); + } + }; + + getMetrics = async (): Promise => { + const rows = await this.jobsRepository.findAll(); + const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); + const now = Date.now(); + return rows.reduce( + (acc, job) => { + acc.jobs++; + acc.totalRuns += job.runCount; + acc.totalFailures += job.failCount; + if (job.lockedBy !== null && job.lockedUntil !== null && job.lockedUntil > now) acc.activeJobs++; + if (job.failCount > 0) { + acc.failingJobs++; + acc.jobsWithFailures.push({ + monitorId: job.refId ?? job.id, + monitorUrl: null, + monitorType: job.type, + failedAt: job.lastFinishedAt, + failCount: job.failCount, + failReason: job.lastFailReason, + }); + } + return acc; + }, + { + jobs: 0, + activeJobs: 0, + failingJobs: 0, + jobsWithFailures: [], + totalRuns: 0, + totalFailures: 0, + workers, + } + ); + }; + + getJobs = async (pagination: WorkerJobsPagination) => { + const { jobs, count } = await this.jobsRepository.findPage(pagination); + return { jobs: jobs.map(this.toSummary), count }; + }; + + flushQueues = async () => { + await this.shutdown(); + return { success: true }; + }; + + shutdown = async () => { + this.stopped = true; + // Stop all timers + for (const timer of this.timers.values()) { + clearTimeout(timer); + } + this.timers.clear(); // Clear the map out + // Remove workers from registry + await this.queueWorkersRepository.deleteById(this.workerId); + }; +} From f0a5974eaf787474919872301ae32cdc326d6c1d Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 23 Jun 2026 11:31:35 -0700 Subject: [PATCH 2/6] update tests --- .../unit/services/maintenanceWindowService.test.ts | 6 +++--- server/test/unit/services/monitorService.test.ts | 4 ++-- server/test/unit/services/userService.test.ts | 12 ++++++------ 3 files changed, 11 insertions(+), 11 deletions(-) diff --git a/server/test/unit/services/maintenanceWindowService.test.ts b/server/test/unit/services/maintenanceWindowService.test.ts index 5c7387da8..9547eaac8 100644 --- a/server/test/unit/services/maintenanceWindowService.test.ts +++ b/server/test/unit/services/maintenanceWindowService.test.ts @@ -3,7 +3,7 @@ import { MaintenanceWindowService } from "../../../src/domain/maintenance-window import type { IMaintenanceWindowsRepository } from "../../../src/domain/maintenance-windows/maintenance-window.repository.interface.ts"; import type { IMonitorsRepository } from "../../../src/domain/monitors/monitor.repository.interface.ts"; import type { IJobsRepository } from "../../../src/domain/jobs/job.repository.interface.ts"; -import type { IWorker } from "../../../src/worker/worker.interface.ts"; +import type { IJobScheduler } from "../../../src/worker/worker.interface.ts"; import type { MaintenanceWindow } from "../../../src/domain/maintenance-windows/maintenance-window.type.ts"; // ── Helpers ────────────────────────────────────────────────────────────────── @@ -37,7 +37,7 @@ const createJobsRepo = () => const createWorker = () => ({ wake: jest.fn(), - }) as unknown as jest.Mocked; + }) as unknown as jest.Mocked; const createService = (overrides?: { monitorsRepository?: ReturnType; @@ -49,7 +49,7 @@ const createService = (overrides?: { const maintenanceWindowsRepository = overrides?.maintenanceWindowsRepository ?? createMaintenanceWindowsRepo(); const jobsRepository = overrides?.jobsRepository ?? createJobsRepo(); const worker = overrides?.worker ?? createWorker(); - const service = new MaintenanceWindowService({ monitorsRepository, maintenanceWindowsRepository, jobsRepository, worker }); + const service = new MaintenanceWindowService({ monitorsRepository, maintenanceWindowsRepository, jobsRepository, scheduler: worker }); return { service, monitorsRepository, maintenanceWindowsRepository, jobsRepository, worker }; }; diff --git a/server/test/unit/services/monitorService.test.ts b/server/test/unit/services/monitorService.test.ts index d64007ebe..65d6923a6 100644 --- a/server/test/unit/services/monitorService.test.ts +++ b/server/test/unit/services/monitorService.test.ts @@ -98,7 +98,7 @@ const createService = ( const jobQueue = overrides.jobQueue ?? createJobQueueMock(); const logger = overrides.logger ?? createMockLogger(); const service = new MonitorService({ - worker: jobQueue as any, + scheduler: jobQueue as any, logger: logger as any, games: (overrides.games ?? { cs2: { name: "Counter-Strike 2" } }) as any, monitorsRepository: overrides.monitorsRepository ?? createMonitorsRepositoryMock(), @@ -1535,7 +1535,7 @@ describe("MonitorService", () => { const logger = createMockLogger(); const service = new MonitorService({ - worker: jobQueue as any, + scheduler: jobQueue as any, logger: logger as any, games: {} as any, monitorsRepository, diff --git a/server/test/unit/services/userService.test.ts b/server/test/unit/services/userService.test.ts index 7b2fb380b..061003e7f 100644 --- a/server/test/unit/services/userService.test.ts +++ b/server/test/unit/services/userService.test.ts @@ -65,7 +65,7 @@ const createService = (overrides?: Record) => { const settingsService = { getSettings: jest.fn().mockReturnValue(makeAppSettings()), }; - const worker = { + const scheduler = { deleteJob: jest.fn().mockResolvedValue(undefined), }; const jwtMock = { @@ -85,7 +85,7 @@ const createService = (overrides?: Record) => { monitorsRepository, emailService, settingsService, - worker, + scheduler, jwt: jwtMock, crypto: cryptoMock, ...overrides, @@ -475,23 +475,23 @@ describe("UserService", () => { describe("deleteUser", () => { it("deletes user and cleans up monitor jobs for superadmin", async () => { - const { service, usersRepository, worker, monitorsRepository } = createService(); + const { service, usersRepository, scheduler, monitorsRepository } = createService(); await service.deleteUser({ userId: "user-1", teamId: "team-1", roles: ["superadmin"] }); expect(monitorsRepository.findByTeamId).toHaveBeenCalledWith("team-1", {}); - expect(worker.deleteJob).toHaveBeenCalledTimes(2); + expect(scheduler.deleteJob).toHaveBeenCalledTimes(2); expect(usersRepository.deleteById).toHaveBeenCalledWith("user-1"); }); it("handles null monitors for superadmin gracefully", async () => { - const { service, usersRepository, worker } = createService({ + const { service, usersRepository, scheduler } = createService({ monitorsRepository: { findByTeamId: jest.fn().mockResolvedValue(null) }, }); await service.deleteUser({ userId: "user-1", teamId: "team-1", roles: ["superadmin"] }); - expect(worker.deleteJob).not.toHaveBeenCalled(); + expect(scheduler.deleteJob).not.toHaveBeenCalled(); expect(usersRepository.deleteById).toHaveBeenCalledWith("user-1"); }); From f59a7c31132f889186afcba76a1961f1464de499 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 23 Jun 2026 12:26:42 -0700 Subject: [PATCH 3/6] extract shared services --- server/src/config/services.api.ts | 0 server/src/config/services.shared.ts | 218 +++++++++++++++++++++++++++ server/src/config/services.ts | 154 +++++-------------- server/src/config/services.worker.ts | 0 4 files changed, 253 insertions(+), 119 deletions(-) create mode 100644 server/src/config/services.api.ts create mode 100644 server/src/config/services.shared.ts create mode 100644 server/src/config/services.worker.ts diff --git a/server/src/config/services.api.ts b/server/src/config/services.api.ts new file mode 100644 index 000000000..e69de29bb diff --git a/server/src/config/services.shared.ts b/server/src/config/services.shared.ts new file mode 100644 index 000000000..49b73f52e --- /dev/null +++ b/server/src/config/services.shared.ts @@ -0,0 +1,218 @@ +import { Mongoose } from "mongoose"; +import { hostname } from "node:os"; +import { randomUUID } from "node:crypto"; +import fs from "fs"; +import path from "path"; +import nodemailer from "nodemailer"; +import pkg from "handlebars"; +const { compile } = pkg; +import mjml2html from "mjml"; + +import MongoDB from "@/db/db.mongo.js"; +import { IDb } from "@/db/db.interface.js"; +import { ISettingsService, EnvConfig, SettingsService } from "@/domain/app-settings/app-settings.service.js"; +import { ICheckService, CheckService } from "@/domain/checks/check.service.js"; +import { IGeoChecksService, GeoChecksService } from "@/domain/geo-checks/geo-check.service.js"; +import { IIncidentService, IncidentService } from "@/domain/incidents/incident.service.js"; +import { INotificationMessageBuilder, NotificationMessageBuilder } from "@/domain/notifications/notification.message-builder.js"; +import { INotificationsService, NotificationsService } from "@/domain/notifications/notification.service.js"; +import { IEmailService, EmailService } from "@/service/emailService.js"; +import { GlobalPingService } from "@/service/globalPingService.js"; +import { ILogger } from "@/utils/logger.js"; + +// Notification providers +import { WebhookProvider } from "@/domain/notifications/providers/webhook.js"; +import { SlackProvider } from "@/domain/notifications/providers/slack.js"; +import { EmailProvider } from "@/domain/notifications/providers/email.js"; +import { DiscordProvider } from "@/domain/notifications/providers/discord.js"; +import { PagerDutyProvider } from "@/domain/notifications/providers/pagerduty.js"; +import { MatrixProvider } from "@/domain/notifications/providers/matrix.js"; +import { TeamsProvider } from "@/domain/notifications/providers/teams.js"; +import { TelegramProvider } from "@/domain/notifications/providers/telegram.js"; +import { PushoverProvider } from "@/domain/notifications/providers/pushover.js"; +import { TwilioProvider } from "@/domain/notifications/providers/twilio.js"; +import { NtfyProvider } from "@/domain/notifications/providers/ntfy.js"; + +// Repository interfaces +import { ISettingsRepository } from "@/domain/app-settings/app-settings-repository.interface.js"; +import { IChecksRepository } from "@/domain/checks/check.repository.interface.js"; +import { IGeoChecksRepository } from "@/domain/geo-checks/geo-check.repository.interface.js"; +import { IIncidentsRepository } from "@/domain/incidents/incident.repository.interface.js"; +import { IInvitesRepository } from "@/domain/invites/invite.repository.interface.js"; +import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; +import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js"; +import { IMonitorStatsRepository } from "@/domain/monitor-stats/monitor-stats.repository.interface.js"; +import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; +import { INotificationsRepository } from "@/domain/notifications/notification.repository.interface.js"; +import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js"; +import { IRecoveryTokensRepository } from "@/domain/recovery-tokens/recovery-token.repository.interface.js"; +import { IStatusPagesRepository } from "@/domain/status-pages/status-page-repository.interface.js"; +import { ITagsRepository } from "@/domain/tags/tag.repository.interface.js"; +import { ITeamsRepository } from "@/domain/teams/team.repository.interface.js"; +import { IUsersRepository } from "@/domain/users/user.repository.interface.js"; + +// Mongo repository implementations +import MongoSettingsRepository from "@/domain/app-settings/app-settings.repository.mongo.js"; +import MongoChecksRepository from "@/domain/checks/check.repository.mongo.js"; +import MongoGeoChecksRepository from "@/domain/geo-checks/geo-check.repository.mongo.js"; +import MongoIncidentsRepository from "@/domain/incidents/incident.repository.mongo.js"; +import MongoInvitesRepository from "@/domain/invites/invite.repository.mongo.js"; +import MongoJobsRepository from "@/domain/jobs/job.repository.mongo.js"; +import MongoMaintenanceWindowsRepository from "@/domain/maintenance-windows/maintenance-window.repository.mongo.js"; +import MongoMonitorStatsRepository from "@/domain/monitor-stats/monitor-stats.repository.mongo.js"; +import MongoMonitorsRepository from "@/domain/monitors/monitor.repository.mongo.js"; +import MongoNotificationsRepository from "@/domain/notifications/notification.repository.mongo.js"; +import MongoQueueWorkersRepository from "@/domain/queue-workers/queue-worker.repository.mongo.js"; +import MongoRecoveryTokensRepository from "@/domain/recovery-tokens/recovery-token.repository.mongo.js"; +import MongoStatusPagesRepository from "@/domain/status-pages/status-page-repository.mongo.js"; +import MongoTagsRepository from "@/domain/tags/tag.repository.mongo.js"; +import MongoTeamsRepository from "@/domain/teams/team.repository.model.js"; +import MongoUsersRepository from "@/domain/users/user.repository.mongo.js"; + +// Shared infrastructure + business services that both the API and the worker process construct. +export interface SharedServices { + logger: ILogger; + db: IDb; + settingsService: ISettingsService; + emailService: IEmailService; + notificationMessageBuilder: INotificationMessageBuilder; + incidentService: IIncidentService; + checkService: ICheckService; + geoChecksService: IGeoChecksService; + notificationsService: INotificationsService; + + // Queue identity (one per process; the worker reuses these) + workerId: string; + jobsRepository: IJobsRepository; + queueWorkersRepository: IQueueWorkersRepository; + + // Repositories + monitorsRepository: IMonitorsRepository; + checksRepository: IChecksRepository; + geoChecksRepository: IGeoChecksRepository; + monitorStatsRepository: IMonitorStatsRepository; + statusPagesRepository: IStatusPagesRepository; + usersRepository: IUsersRepository; + invitesRepository: IInvitesRepository; + recoveryTokensRepository: IRecoveryTokensRepository; + settingsRepository: ISettingsRepository; + notificationsRepository: INotificationsRepository; + tagsRepository: ITagsRepository; + incidentsRepository: IIncidentsRepository; + teamsRepository: ITeamsRepository; + maintenanceWindowsRepository: IMaintenanceWindowsRepository; +} + +export const buildShared = async ({ + logger, + envSettings, + settingsService, +}: { + logger: ILogger; + envSettings: EnvConfig; + settingsService: ISettingsService; +}): Promise => { + // Shared worker ID + const workerId = `${hostname()}:${process.pid}:${randomUUID()}`; + + // Create DB + let db: IDb | null = null; + db = new MongoDB(logger, envSettings); + await db.connect(); + + // Repositories + const monitorsRepository = new MongoMonitorsRepository(); + const checksRepository = new MongoChecksRepository(logger); + const incidentsRepository = new MongoIncidentsRepository(); + const usersRepository = new MongoUsersRepository(); + const geoChecksRepository = new MongoGeoChecksRepository(logger); + const notificationsRepository = new MongoNotificationsRepository(); + const jobsRepository = new MongoJobsRepository(workerId); + const queueWorkersRepository = new MongoQueueWorkersRepository(); + const monitorStatsRepository = new MongoMonitorStatsRepository(); + const statusPagesRepository = new MongoStatusPagesRepository(); + const invitesRepository = new MongoInvitesRepository(); + const recoveryTokensRepository = new MongoRecoveryTokensRepository(); + const settingsRepository = new MongoSettingsRepository(); + const tagsRepository = new MongoTagsRepository(); + const teamsRepository = new MongoTeamsRepository(); + const maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository(); + + // Inject settings repository into settings service (now that DB is connected) + (settingsService as SettingsService).setRepository(settingsRepository); + + // Services + const notificationMessageBuilder = new NotificationMessageBuilder(); + const emailService = new EmailService(settingsService, fs, path, compile, mjml2html, nodemailer, logger); + const incidentService = new IncidentService(logger, incidentsRepository, monitorsRepository, usersRepository, notificationMessageBuilder); + const checkService = new CheckService(monitorsRepository, logger, checksRepository); + + const globalPingService = new GlobalPingService(logger); + const geoChecksService = new GeoChecksService({ + logger, + geoChecksRepository, + globalPingService, + monitorsRepository, + }); + + const webhookProvider = new WebhookProvider(logger); + const slackProvider = new SlackProvider(logger); + const emailProvider = new EmailProvider(emailService, logger); + const discordProvider = new DiscordProvider(logger); + const pagerDutyProvider = new PagerDutyProvider(logger); + const matrixProvider = new MatrixProvider(logger); + const teamsProvider = new TeamsProvider(logger); + const telegramProvider = new TelegramProvider(logger); + const pushoverProvider = new PushoverProvider(logger); + const twilioProvider = new TwilioProvider(logger); + const ntfyProvider = new NtfyProvider(logger); + + const notificationsService = new NotificationsService( + notificationsRepository, + monitorsRepository, + webhookProvider, + emailProvider, + slackProvider, + discordProvider, + pagerDutyProvider, + matrixProvider, + teamsProvider, + telegramProvider, + pushoverProvider, + twilioProvider, + ntfyProvider, + settingsService, + logger, + notificationMessageBuilder + ); + + const sharedServices: SharedServices = { + logger, + db, + settingsService, + emailService, + notificationMessageBuilder, + incidentService, + checkService, + geoChecksService, + notificationsService, + workerId, + jobsRepository, + monitorsRepository, + queueWorkersRepository, + checksRepository, + geoChecksRepository, + monitorStatsRepository, + statusPagesRepository, + usersRepository, + invitesRepository, + recoveryTokensRepository, + settingsRepository, + notificationsRepository, + tagsRepository, + incidentsRepository, + teamsRepository, + maintenanceWindowsRepository, + }; + return sharedServices; +}; diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 7bc59f881..78dabadf8 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -1,35 +1,21 @@ import { Mongoose } from "mongoose"; -import { hostname } from "node:os"; -import { randomUUID } from "node:crypto"; -import MongoDB from "../db/db.mongo.js"; +import { buildShared } from "@/config/services.shared.js"; import { IDb } from "@/db/db.interface.js"; -import { EnvConfig, ISettingsService, SettingsService } from "@/domain/app-settings/app-settings.service.js"; -import { CheckService, ICheckService } from "@/domain/checks/check.service.js"; +import { EnvConfig, ISettingsService } from "@/domain/app-settings/app-settings.service.js"; +import { ICheckService } from "@/domain/checks/check.service.js"; import { DiagnosticService, IDiagnosticService } from "@/domain/diagnostics/diagnostic.service.js"; -import { GeoChecksService, IGeoChecksService } from "@/domain/geo-checks/geo-check.service.js"; -import { IIncidentService, IncidentService } from "@/domain/incidents/incident.service.js"; +import { IGeoChecksService } from "@/domain/geo-checks/geo-check.service.js"; +import { IIncidentService } from "@/domain/incidents/incident.service.js"; import { IInviteService, InviteService } from "@/domain/invites/invite.service.js"; import { IMaintenanceWindowService, MaintenanceWindowService } from "@/domain/maintenance-windows/maintenance-window.service.js"; import { IMonitorService, MonitorService } from "@/domain/monitors/monitor.service.js"; -import { INotificationMessageBuilder, NotificationMessageBuilder } from "@/domain/notifications/notification.message-builder.js"; -import { INotificationsService, NotificationsService } from "@/domain/notifications/notification.service.js"; -import { DiscordProvider } from "@/domain/notifications/providers/discord.js"; -import { EmailProvider } from "@/domain/notifications/providers/email.js"; -import { MatrixProvider } from "@/domain/notifications/providers/matrix.js"; -import { NtfyProvider } from "@/domain/notifications/providers/ntfy.js"; -import { PagerDutyProvider } from "@/domain/notifications/providers/pagerduty.js"; -import { PushoverProvider } from "@/domain/notifications/providers/pushover.js"; -import { SlackProvider } from "@/domain/notifications/providers/slack.js"; -import { TeamsProvider } from "@/domain/notifications/providers/teams.js"; -import { TelegramProvider } from "@/domain/notifications/providers/telegram.js"; -import { TwilioProvider } from "@/domain/notifications/providers/twilio.js"; -import { WebhookProvider } from "@/domain/notifications/providers/webhook.js"; +import { INotificationMessageBuilder } from "@/domain/notifications/notification.message-builder.js"; +import { INotificationsService } from "@/domain/notifications/notification.service.js"; import { IStatusPageService, StatusPageService } from "@/domain/status-pages/status-page.service.js"; import { ITagsService, TagsService } from "@/domain/tags/tag.service.js"; 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 { IEmailService } from "@/service/emailService.js"; import { WorkerHelper } from "@/worker/worker.helper.js"; import { IQueueWorker } from "@/worker/worker.interface.js"; import { INetworkService, NetworkService } from "@/service/networkService.js"; @@ -56,12 +42,6 @@ import got from "got"; import ping from "ping"; import Docker from "dockerode"; import net from "net"; -import fs from "fs"; -import path from "path"; -import nodemailer from "nodemailer"; -import pkg from "handlebars"; -const { compile } = pkg; -import mjml2html from "mjml"; import jwt from "jsonwebtoken"; import crypto from "crypto"; import { games, GameDig } from "gamedig"; @@ -72,33 +52,19 @@ import WebSocket from "ws"; // Repositories import { ISettingsRepository } from "@/domain/app-settings/app-settings-repository.interface.js"; -import MongoSettingsRepository from "@/domain/app-settings/app-settings.repository.mongo.js"; import { IChecksRepository } from "@/domain/checks/check.repository.interface.js"; -import MongoChecksRepository from "@/domain/checks/check.repository.mongo.js"; import { IGeoChecksRepository } from "@/domain/geo-checks/geo-check.repository.interface.js"; -import MongoGeoChecksRepository from "@/domain/geo-checks/geo-check.repository.mongo.js"; import { IIncidentsRepository } from "@/domain/incidents/incident.repository.interface.js"; -import MongoIncidentsRepository from "@/domain/incidents/incident.repository.mongo.js"; import { IInvitesRepository } from "@/domain/invites/invite.repository.interface.js"; -import MongoInvitesRepository from "@/domain/invites/invite.repository.mongo.js"; import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js"; -import MongoMaintenanceWindowsRepository from "@/domain/maintenance-windows/maintenance-window.repository.mongo.js"; import { IMonitorStatsRepository } from "@/domain/monitor-stats/monitor-stats.repository.interface.js"; -import MongoMonitorStatsRepository from "@/domain/monitor-stats/monitor-stats.repository.mongo.js"; import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; -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 { 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"; -import MongoStatusPagesRepository from "@/domain/status-pages/status-page-repository.mongo.js"; import { ITagsRepository } from "@/domain/tags/tag.repository.interface.js"; -import MongoTagsRepository from "@/domain/tags/tag.repository.mongo.js"; import { ITeamsRepository } from "@/domain/teams/team.repository.interface.js"; -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 { NotificationReactor } from "@/worker/reactors/reactor.notification.js"; import { ReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js"; @@ -107,8 +73,6 @@ import { GeoChecksPipeline } from "@/worker/worker.check-pipeline.js"; import { CheckProducer } from "@/worker/worker.check-producer.js"; import { CheckEvaluator } from "@/worker/worker.check-evaluator.js"; import { DBQueueWorker } from "@/worker/worker.db-queue.js"; -import MongoJobsRepository from "@/domain/jobs/job.repository.mongo.js"; -import MongoQueueWorkersRepository from "@/domain/queue-workers/queue-worker.repository.mongo.js"; export type InitializedServices = { settingsService: ISettingsService; @@ -158,33 +122,33 @@ export const initializeServices = async ({ envSettings: EnvConfig; settingsService: ISettingsService; }): Promise => { - // Create DB - - let db: IDb | null = null; - db = new MongoDB(logger, envSettings); - await db.connect(); - - const monitorsRepository = new MongoMonitorsRepository(); - const checksRepository = new MongoChecksRepository(logger); - const geoChecksRepository = new MongoGeoChecksRepository(logger); - const monitorStatsRepository = new MongoMonitorStatsRepository(); - const statusPagesRepository = new MongoStatusPagesRepository(); - const usersRepository = new MongoUsersRepository(); - const invitesRepository = new MongoInvitesRepository(); - const recoveryTokensRepository = new MongoRecoveryTokensRepository(); - const settingsRepository = new MongoSettingsRepository(); - const notificationsRepository = new MongoNotificationsRepository(); - const tagsRepository = new MongoTagsRepository(); - const incidentsRepository = new MongoIncidentsRepository(); - const teamsRepository = new MongoTeamsRepository(); - const maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository(); - // One identity per process - const workerId = `${hostname()}:${process.pid}:${randomUUID()}`; - const jobsRepository = new MongoJobsRepository(workerId); - const queueWorkersRepository = new MongoQueueWorkersRepository(); - - // Inject settings repository into settings service (now that DB is connected) - (settingsService as SettingsService).setRepository(settingsRepository); + const shared = await buildShared({ logger, envSettings, settingsService }); + const { + db, + emailService, + notificationMessageBuilder, + incidentService, + checkService, + geoChecksService, + notificationsService, + workerId, + jobsRepository, + monitorsRepository, + queueWorkersRepository, + checksRepository, + geoChecksRepository, + monitorStatsRepository, + statusPagesRepository, + usersRepository, + invitesRepository, + recoveryTokensRepository, + settingsRepository, + notificationsRepository, + tagsRepository, + incidentsRepository, + teamsRepository, + maintenanceWindowsRepository, + } = shared; // Network providers const pingProvider = new PingProvider(ping); @@ -210,59 +174,11 @@ export const initializeServices = async ({ webSocketProvider, dnsProvider, ]); - const emailService = new EmailService(settingsService, fs, path, compile, mjml2html, nodemailer, logger); - - const notificationMessageBuilder = new NotificationMessageBuilder(); - - const incidentService = new IncidentService(logger, incidentsRepository, monitorsRepository, usersRepository, notificationMessageBuilder); - - const checkService = new CheckService(monitorsRepository, logger, checksRepository); - - const globalPingService = new GlobalPingService(logger); - - const geoChecksService = new GeoChecksService({ - logger, - geoChecksRepository, - globalPingService, - monitorsRepository, - }); const bufferService = new BufferService(logger, checkService, geoChecksService, settingsService, jobsRepository); const statusService = new StatusService(logger, monitorsRepository, monitorStatsRepository); - // Notification providers - const webhookProvider = new WebhookProvider(logger); - const slackProvider = new SlackProvider(logger); - const emailProvider = new EmailProvider(emailService, logger); - const discordProvider = new DiscordProvider(logger); - const pagerDutyProvider = new PagerDutyProvider(logger); - const matrixProvider = new MatrixProvider(logger); - const teamsProvider = new TeamsProvider(logger); - const telegramProvider = new TelegramProvider(logger); - const pushoverProvider = new PushoverProvider(logger); - const twilioProvider = new TwilioProvider(logger); - const ntfyProvider = new NtfyProvider(logger); - - const notificationsService = new NotificationsService( - notificationsRepository, - monitorsRepository, - webhookProvider, - emailProvider, - slackProvider, - discordProvider, - pagerDutyProvider, - matrixProvider, - teamsProvider, - telegramProvider, - pushoverProvider, - twilioProvider, - ntfyProvider, - settingsService, - logger, - notificationMessageBuilder - ); - const tagsService = new TagsService(tagsRepository, monitorsRepository); const monitorStatusPolicy = new MonitorStatusPolicy(); diff --git a/server/src/config/services.worker.ts b/server/src/config/services.worker.ts new file mode 100644 index 000000000..e69de29bb From 71d3e789ff3e06237e1eff7b512b8c3d60cf5bac Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 24 Jun 2026 10:30:21 -0700 Subject: [PATCH 4/6] split services to api, worker, and shared --- server/openapi.json | 24 +- server/src/app.ts | 6 +- server/src/config/controllers.ts | 32 +-- server/src/config/routes.ts | 8 +- server/src/config/services.api.ts | 109 +++++++++ server/src/config/services.ts | 309 ------------------------ server/src/config/services.worker.ts | 151 ++++++++++++ server/src/domain/checks/check.model.ts | 1 + server/src/index.ts | 78 ++++-- server/src/shutdown.ts | 10 +- 10 files changed, 375 insertions(+), 353 deletions(-) delete mode 100644 server/src/config/services.ts diff --git a/server/openapi.json b/server/openapi.json index f3a6a2505..d0bbe45dc 100644 --- a/server/openapi.json +++ b/server/openapi.json @@ -386,6 +386,10 @@ "type": "string", "enum": ["equal", "include", "regex"] }, + "method": { + "type": "string", + "enum": ["GET", "HEAD"] + }, "notifications": { "type": "array", "items": { @@ -466,6 +470,9 @@ }, "updatedAt": { "type": "string" + }, + "lastEvaluatedAt": { + "type": "number" } }, "required": [ @@ -480,6 +487,7 @@ "statusWindowThreshold", "ignoreTlsErrors", "useAdvancedMatching", + "method", "notifications", "tags", "cpuAlertThreshold", @@ -493,7 +501,8 @@ "teamId", "userId", "createdAt", - "updatedAt" + "updatedAt", + "lastEvaluatedAt" ], "additionalProperties": { "nullable": true @@ -7066,6 +7075,10 @@ } ] }, + "method": { + "type": "string", + "enum": ["GET", "HEAD"] + }, "gameId": { "anyOf": [ { @@ -7666,6 +7679,11 @@ } ] }, + "method": { + "type": "string", + "enum": ["GET", "HEAD"], + "default": "GET" + }, "url": { "type": "string", "minLength": 1 @@ -8278,6 +8296,10 @@ } ] }, + "method": { + "type": "string", + "enum": ["GET", "HEAD"] + }, "port": { "type": "number", "minimum": 1, diff --git a/server/src/app.ts b/server/src/app.ts index 907c0f77d..d559edb1f 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -9,20 +9,20 @@ import { handleErrors } from "@/api/middleware/handleErrors.js"; import { generalApiLimiter } from "@/api/middleware/rateLimiter.js"; import { sanitizeBody, sanitizeQuery } from "@/api/middleware/sanitization.js"; import { setupRoutes } from "@/config/routes.js"; -import { InitializedServices } from "@/config/services.js"; import { InitializedControllers } from "@/config/controllers.js"; import { EnvConfig } from "@/domain/app-settings/app-settings.service.js"; import { createStatusPageCorsOrigin } from "@/api/middleware/statusPageCorsOrigin.js"; import { isPublicStatusPageApiPath } from "@/api/middleware/statusPagePublicApiPath.js"; +import { ApiServices } from "@/config/services.api.js"; export const createApp = ({ - services, + apiServices: services, controllers, envSettings, frontendPath, openApiSpec, }: { - services: InitializedServices; + apiServices: ApiServices; controllers: InitializedControllers; envSettings: EnvConfig; frontendPath: string; diff --git a/server/src/config/controllers.ts b/server/src/config/controllers.ts index cdbb876b8..7646b41db 100644 --- a/server/src/config/controllers.ts +++ b/server/src/config/controllers.ts @@ -12,7 +12,7 @@ import NotificationController from "../api/controllers/notificationController.js import TagsController from "../api/controllers/tagController.js"; import DiagnosticController from "../api/controllers/diagnosticController.js"; import IncidentController from "../api/controllers/incidentController.js"; -import type { InitializedServices } from "@/config/services.js"; +import { ApiServices } from "@/config/services.api.js"; export interface InitializedControllers { authController: AuthController; @@ -30,21 +30,21 @@ export interface InitializedControllers { diagnosticController: DiagnosticController; incidentController: IncidentController; } -export const initializeControllers = (services: InitializedServices): InitializedControllers => { +export const initializeControllers = (apiServices: ApiServices): InitializedControllers => { return { - authController: new AuthController(services.userService), - monitorController: new MonitorController(services.monitorService, services.notificationsService), - settingsController: new SettingsController(services.settingsService, services.emailService), - checkController: new CheckController(services.checkService), - geoCheckController: new GeoCheckController(services.geoChecksService), - inviteController: new InviteController(services.inviteService), - maintenanceWindowController: new MaintenanceWindowController(services.maintenanceWindowService), - 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), - tagController: new TagsController(services.tagsService), - diagnosticController: new DiagnosticController(services.diagnosticService), - incidentController: new IncidentController(services.incidentService), + authController: new AuthController(apiServices.userService), + monitorController: new MonitorController(apiServices.monitorService, apiServices.notificationsService), + settingsController: new SettingsController(apiServices.settingsService, apiServices.emailService), + checkController: new CheckController(apiServices.checkService), + geoCheckController: new GeoCheckController(apiServices.geoChecksService), + inviteController: new InviteController(apiServices.inviteService), + maintenanceWindowController: new MaintenanceWindowController(apiServices.maintenanceWindowService), + queueController: new QueueController(apiServices.worker), + logController: new LogController(apiServices.logger), + statusPageController: new StatusPageController(apiServices.statusPageService, apiServices.monitorsRepository, apiServices.settingsService), + notificationController: new NotificationController(apiServices.notificationsService, apiServices.monitorsRepository), + tagController: new TagsController(apiServices.tagsService), + diagnosticController: new DiagnosticController(apiServices.diagnosticService), + incidentController: new IncidentController(apiServices.incidentService), }; }; diff --git a/server/src/config/routes.ts b/server/src/config/routes.ts index d3fbab2c7..855c18abb 100644 --- a/server/src/config/routes.ts +++ b/server/src/config/routes.ts @@ -3,7 +3,7 @@ import { createVerifyJWT } from "../api/middleware/verifyJWT.js"; import { createVerifyStatusPageAccess } from "../api/middleware/verifyStatusPageAccess.js"; import { authApiLimiter } from "../api/middleware/rateLimiter.js"; import type { InitializedControllers } from "./controllers.js"; -import type { InitializedServices } from "./services.js"; +import type { ApiServices } from "@/config/services.api.js"; import AuthRoutes from "../api/routes/authRoute.js"; import InviteRoutes from "../api/routes/inviteRoute.js"; @@ -21,8 +21,8 @@ import TagRoutes from "../api/routes/tagRoutes.js"; import IncidentRoutes from "../api/routes/incidentRoute.js"; -export const setupRoutes = (app: Application, controllers: InitializedControllers, services: InitializedServices) => { - const verifyJWT = createVerifyJWT(services.settingsService); +export const setupRoutes = (app: Application, controllers: InitializedControllers, apiServices: ApiServices) => { + const verifyJWT = createVerifyJWT(apiServices.settingsService); const authRoutes = new AuthRoutes(controllers.authController, verifyJWT); const monitorRoutes = new MonitorRoutes(controllers.monitorController); const settingsRoutes = new SettingsRoutes(controllers.settingsController); @@ -32,7 +32,7 @@ export const setupRoutes = (app: Application, controllers: InitializedController const maintenanceWindowRoutes = new MaintenanceWindowRoutes(controllers.maintenanceWindowController); const queueRoutes = new QueueRoutes(controllers.queueController); const logRoutes = new LogRoutes(controllers.logController); - const verifyStatusPageAccess = createVerifyStatusPageAccess(services.statusPagesRepository, verifyJWT); + const verifyStatusPageAccess = createVerifyStatusPageAccess(apiServices.statusPagesRepository, verifyJWT); const statusPageRoutes = new StatusPageRoutes(controllers.statusPageController, verifyJWT, verifyStatusPageAccess); const notificationRoutes = new NotificationRoutes(controllers.notificationController); const tagRoutes = new TagRoutes(controllers.tagController); diff --git a/server/src/config/services.api.ts b/server/src/config/services.api.ts index e69de29bb..0b9361e5d 100644 --- a/server/src/config/services.api.ts +++ b/server/src/config/services.api.ts @@ -0,0 +1,109 @@ +import { DiagnosticService, IDiagnosticService } from "@/domain/diagnostics/diagnostic.service.js"; +import { IInviteService, InviteService } from "@/domain/invites/invite.service.js"; +import { IMaintenanceWindowService, MaintenanceWindowService } from "@/domain/maintenance-windows/maintenance-window.service.js"; +import { IMonitorService, MonitorService } from "@/domain/monitors/monitor.service.js"; +import { IStatusPageService, StatusPageService } from "@/domain/status-pages/status-page.service.js"; +import { ITagsService, TagsService } from "@/domain/tags/tag.service.js"; +import { IUserService, UserService } from "@/domain/users/user.service.js"; +import { IJobScheduler } from "@/worker/worker.interface.js"; + +// Third-party +import jwt from "jsonwebtoken"; +import crypto from "crypto"; +import { games } from "gamedig"; + +import { SharedServices } from "@/config/services.shared.js"; + +export interface ApiServices extends SharedServices { + worker: IJobScheduler; // control-plane handle only (DBQueueWorker in all-in-one, bare JobScheduler in API-only) + userService: IUserService; + monitorService: IMonitorService; + maintenanceWindowService: IMaintenanceWindowService; + inviteService: IInviteService; + statusPageService: IStatusPageService; + tagsService: ITagsService; + diagnosticService: IDiagnosticService; +} + +export const buildApi = (shared: SharedServices, jobScheduler: IJobScheduler): ApiServices => { + const { + logger, + db, + settingsService, + emailService, + monitorsRepository, + checksRepository, + geoChecksRepository, + monitorStatsRepository, + statusPagesRepository, + usersRepository, + invitesRepository, + recoveryTokensRepository, + settingsRepository, + tagsRepository, + incidentsRepository, + teamsRepository, + maintenanceWindowsRepository, + jobsRepository, + } = shared; + + const userService = new UserService({ + crypto, + emailService, + settingsService, + logger, + jwt, + scheduler: jobScheduler, + monitorsRepository, + usersRepository, + invitesRepository, + recoveryTokensRepository, + settingsRepository, + teamsRepository, + }); + + // *********************** + // Business services + // *********************** + + const monitorService = new MonitorService({ + scheduler: jobScheduler, + logger, + games, + monitorsRepository, + checksRepository, + geoChecksRepository, + monitorStatsRepository, + statusPagesRepository, + incidentsRepository, + }); + + const maintenanceWindowService = new MaintenanceWindowService({ + monitorsRepository, + maintenanceWindowsRepository, + jobsRepository, + scheduler: jobScheduler, + }); + + const inviteService = new InviteService({ + invitesRepository, + settingsService, + emailService, + }); + + const statusPageService = new StatusPageService(statusPagesRepository, settingsService); + const tagsService = new TagsService(tagsRepository, monitorsRepository); + const diagnosticService = new DiagnosticService(db); + + return { + ...shared, + worker: jobScheduler, + userService, + monitorService, + maintenanceWindowService, + inviteService, + statusPageService, + tagsService, + diagnosticService, + }; +}; diff --git a/server/src/config/services.ts b/server/src/config/services.ts deleted file mode 100644 index 78dabadf8..000000000 --- a/server/src/config/services.ts +++ /dev/null @@ -1,309 +0,0 @@ -import { Mongoose } from "mongoose"; -import { buildShared } from "@/config/services.shared.js"; -import { IDb } from "@/db/db.interface.js"; -import { EnvConfig, ISettingsService } from "@/domain/app-settings/app-settings.service.js"; -import { ICheckService } from "@/domain/checks/check.service.js"; -import { DiagnosticService, IDiagnosticService } from "@/domain/diagnostics/diagnostic.service.js"; -import { IGeoChecksService } from "@/domain/geo-checks/geo-check.service.js"; -import { IIncidentService } from "@/domain/incidents/incident.service.js"; -import { IInviteService, InviteService } from "@/domain/invites/invite.service.js"; -import { IMaintenanceWindowService, MaintenanceWindowService } from "@/domain/maintenance-windows/maintenance-window.service.js"; -import { IMonitorService, MonitorService } from "@/domain/monitors/monitor.service.js"; -import { INotificationMessageBuilder } from "@/domain/notifications/notification.message-builder.js"; -import { INotificationsService } from "@/domain/notifications/notification.service.js"; -import { IStatusPageService, StatusPageService } from "@/domain/status-pages/status-page.service.js"; -import { ITagsService, TagsService } from "@/domain/tags/tag.service.js"; -import { IUserService, UserService } from "@/domain/users/user.service.js"; -import { BufferService, IBufferService } from "@/service/bufferService.js"; -import { IEmailService } from "@/service/emailService.js"; -import { WorkerHelper } from "@/worker/worker.helper.js"; -import { IQueueWorker } from "@/worker/worker.interface.js"; -import { INetworkService, NetworkService } from "@/service/networkService.js"; -import { IStatusService, StatusService } from "@/service/statusService.js"; -import { MonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js"; - -// Network providers -import { PingProvider } from "@/service/network/PingProvider.js"; -import { HttpProvider } from "@/service/network/HttpProvider.js"; -import { AdvancedMatcher } from "@/service/network/AdvancedMatcher.js"; -import { PageSpeedProvider } from "@/service/network/PageSpeedProvider.js"; -import { HardwareProvider } from "@/service/network/HardwareProvider.js"; -import { DockerProvider } from "@/service/network/DockerProvider.js"; -import { PortProvider } from "@/service/network/PortProvider.js"; -import { GameProvider } from "@/service/network/GameProvider.js"; -import { GrpcProvider } from "@/service/network/GrpcProvider.js"; -import { WebSocketProvider } from "@/service/network/WebSocketProvider.js"; -import { DNSProvider } from "@/service/network/DNSProvider.js"; - -// Third-party -import { Resolver } from "dns/promises"; -import axios from "axios"; -import got from "got"; -import ping from "ping"; -import Docker from "dockerode"; -import net from "net"; -import jwt from "jsonwebtoken"; -import crypto from "crypto"; -import { games, GameDig } from "gamedig"; -import jmespath from "jmespath"; -import * as grpc from "@grpc/grpc-js"; -import * as protoLoader from "@grpc/proto-loader"; -import WebSocket from "ws"; - -// Repositories -import { ISettingsRepository } from "@/domain/app-settings/app-settings-repository.interface.js"; -import { IChecksRepository } from "@/domain/checks/check.repository.interface.js"; -import { IGeoChecksRepository } from "@/domain/geo-checks/geo-check.repository.interface.js"; -import { IIncidentsRepository } from "@/domain/incidents/incident.repository.interface.js"; -import { IInvitesRepository } from "@/domain/invites/invite.repository.interface.js"; -import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js"; -import { IMonitorStatsRepository } from "@/domain/monitor-stats/monitor-stats.repository.interface.js"; -import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; -import { INotificationsRepository } from "@/domain/notifications/notification.repository.interface.js"; -import { IRecoveryTokensRepository } from "@/domain/recovery-tokens/recovery-token.repository.interface.js"; -import { IStatusPagesRepository } from "@/domain/status-pages/status-page-repository.interface.js"; -import { ITagsRepository } from "@/domain/tags/tag.repository.interface.js"; -import { ITeamsRepository } from "@/domain/teams/team.repository.interface.js"; -import { IUsersRepository } from "@/domain/users/user.repository.interface.js"; -import { ILogger } from "@/utils/logger.js"; -import { NotificationReactor } from "@/worker/reactors/reactor.notification.js"; -import { ReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js"; -import { IncidentReactor } from "@/worker/reactors/reactor.incident.js"; -import { GeoChecksPipeline } from "@/worker/worker.check-pipeline.js"; -import { CheckProducer } from "@/worker/worker.check-producer.js"; -import { CheckEvaluator } from "@/worker/worker.check-evaluator.js"; -import { DBQueueWorker } from "@/worker/worker.db-queue.js"; - -export type InitializedServices = { - settingsService: ISettingsService; - db: IDb; - networkService: INetworkService; - emailService: IEmailService; - bufferService: IBufferService; - statusService: IStatusService; - worker: IQueueWorker; - userService: IUserService; - checkService: ICheckService; - geoChecksService: IGeoChecksService; - diagnosticService: IDiagnosticService; - inviteService: IInviteService; - maintenanceWindowService: IMaintenanceWindowService; - monitorService: IMonitorService; - incidentService: IIncidentService; - logger: ILogger; - notificationsService: INotificationsService; - tagsService: ITagsService; - statusPageService: IStatusPageService; - notificationMessageBuilder: INotificationMessageBuilder; - - // Repositories - monitorsRepository: IMonitorsRepository; - checksRepository: IChecksRepository; - geoChecksRepository: IGeoChecksRepository; - monitorStatsRepository: IMonitorStatsRepository; - statusPagesRepository: IStatusPagesRepository; - usersRepository: IUsersRepository; - invitesRepository: IInvitesRepository; - recoveryTokensRepository: IRecoveryTokensRepository; - settingsRepository: ISettingsRepository; - notificationsRepository: INotificationsRepository; - tagsRepository: ITagsRepository; - incidentsRepository: IIncidentsRepository; - teamsRepository: ITeamsRepository; - maintenanceWindowsRepository: IMaintenanceWindowsRepository; -}; - -export const initializeServices = async ({ - logger, - envSettings, - settingsService, -}: { - logger: ILogger; - envSettings: EnvConfig; - settingsService: ISettingsService; -}): Promise => { - const shared = await buildShared({ logger, envSettings, settingsService }); - const { - db, - emailService, - notificationMessageBuilder, - incidentService, - checkService, - geoChecksService, - notificationsService, - workerId, - jobsRepository, - monitorsRepository, - queueWorkersRepository, - checksRepository, - geoChecksRepository, - monitorStatsRepository, - statusPagesRepository, - usersRepository, - invitesRepository, - recoveryTokensRepository, - settingsRepository, - notificationsRepository, - tagsRepository, - incidentsRepository, - teamsRepository, - maintenanceWindowsRepository, - } = shared; - - // Network providers - const pingProvider = new PingProvider(ping); - const httpProvider = new HttpProvider(got, new AdvancedMatcher(jmespath)); - const pageSpeedProvider = new PageSpeedProvider(httpProvider, settingsService, logger); - const hardwareProvider = new HardwareProvider(httpProvider); - const dockerProvider = new DockerProvider(logger, Docker); - const portProvider = new PortProvider(net); - const gameProvider = new GameProvider(logger, GameDig); - const grpcProvider = new GrpcProvider(grpc, protoLoader); - const webSocketProvider = new WebSocketProvider(WebSocket); - const dnsProvider = new DNSProvider(() => new Resolver()); - - const networkService = new NetworkService(axios, logger, [ - pingProvider, - httpProvider, - pageSpeedProvider, - hardwareProvider, - dockerProvider, - portProvider, - gameProvider, - grpcProvider, - webSocketProvider, - dnsProvider, - ]); - - const bufferService = new BufferService(logger, checkService, geoChecksService, settingsService, jobsRepository); - - const statusService = new StatusService(logger, monitorsRepository, monitorStatsRepository); - - const tagsService = new TagsService(tagsRepository, monitorsRepository); - - const monitorStatusPolicy = new MonitorStatusPolicy(); - // reactors and dispatcher - const notificationReactor = new NotificationReactor(notificationsService); - const incidentReactor = new IncidentReactor(incidentService); - const reactorDispatcher = new ReactorDispatcher(logger, [notificationReactor, incidentReactor]); - - // Check producer/evaluator - const checkProducer = new CheckProducer(monitorsRepository, maintenanceWindowsRepository, checkService, networkService, bufferService, logger); - const checkEvaluator = new CheckEvaluator(statusService, monitorStatusPolicy); - - // pipelines - const geoCheckPipeline = new GeoChecksPipeline(maintenanceWindowsRepository, geoChecksService, bufferService, logger); - - const workerHelper = new WorkerHelper( - logger, - checkService, - settingsService, - monitorsRepository, - jobsRepository, - teamsRepository, - monitorStatsRepository, - checksRepository, - incidentsRepository, - geoChecksRepository - ); - - const worker = await DBQueueWorker.create( - logger, - jobsRepository, - monitorsRepository, - checksRepository, - checkService, - checkProducer, - checkEvaluator, - geoCheckPipeline, - reactorDispatcher, - workerHelper, - queueWorkersRepository, - envSettings.queueMode, - envSettings.queuePrimaryProcesses, - workerId - ); - - // Business services - const userService = new UserService({ - crypto, - emailService, - settingsService, - logger, - jwt, - scheduler: worker, - monitorsRepository, - usersRepository, - invitesRepository, - recoveryTokensRepository, - settingsRepository, - teamsRepository, - }); - - const diagnosticService = new DiagnosticService(db); - const inviteService = new InviteService({ - invitesRepository, - settingsService, - emailService, - }); - const maintenanceWindowService = new MaintenanceWindowService({ - monitorsRepository, - maintenanceWindowsRepository, - jobsRepository, - scheduler: worker, - }); - const monitorService = new MonitorService({ - scheduler: worker, - logger, - games, - monitorsRepository, - checksRepository, - geoChecksRepository, - monitorStatsRepository, - statusPagesRepository, - incidentsRepository, - }); - - const statusPageService = new StatusPageService(statusPagesRepository, settingsService); - - const services = { - settingsService, - db, - networkService, - emailService, - bufferService, - statusService, - worker, - userService, - checkService, - geoChecksService, - diagnosticService, - inviteService, - maintenanceWindowService, - monitorService, - incidentService, - logger, - notificationsService, - tagsService, - statusPageService, - notificationMessageBuilder, - - // Repositories - monitorsRepository, - checksRepository, - geoChecksRepository, - monitorStatsRepository, - statusPagesRepository, - usersRepository, - invitesRepository, - recoveryTokensRepository, - settingsRepository, - notificationsRepository, - tagsRepository, - incidentsRepository, - teamsRepository, - maintenanceWindowsRepository, - }; - - return services; -}; diff --git a/server/src/config/services.worker.ts b/server/src/config/services.worker.ts index e69de29bb..93233cd37 100644 --- a/server/src/config/services.worker.ts +++ b/server/src/config/services.worker.ts @@ -0,0 +1,151 @@ +import { Resolver } from "dns/promises"; +import axios from "axios"; +import got from "got"; +import ping from "ping"; +import Docker from "dockerode"; +import net from "net"; +import { GameDig } from "gamedig"; +import jmespath from "jmespath"; +import * as grpc from "@grpc/grpc-js"; +import * as protoLoader from "@grpc/proto-loader"; +import WebSocket from "ws"; + +import { EnvConfig } from "@/domain/app-settings/app-settings.service.js"; +import { SharedServices } from "@/config/services.shared.js"; +import { INetworkService, NetworkService } from "@/service/networkService.js"; +import { IBufferService, BufferService } from "@/service/bufferService.js"; +import { IStatusService, StatusService } from "@/service/statusService.js"; +import { IQueueWorker } from "@/worker/worker.interface.js"; +import { MonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js"; +import { WorkerHelper } from "@/worker/worker.helper.js"; +import { CheckProducer } from "@/worker/worker.check-producer.js"; +import { CheckEvaluator } from "@/worker/worker.check-evaluator.js"; +import { GeoChecksPipeline } from "@/worker/worker.check-pipeline.js"; +import { NotificationReactor } from "@/worker/reactors/reactor.notification.js"; +import { IncidentReactor } from "@/worker/reactors/reactor.incident.js"; +import { ReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js"; +import { DBQueueWorker } from "@/worker/worker.db-queue.js"; + +// Network providers +import { PingProvider } from "@/service/network/PingProvider.js"; +import { HttpProvider } from "@/service/network/HttpProvider.js"; +import { AdvancedMatcher } from "@/service/network/AdvancedMatcher.js"; +import { PageSpeedProvider } from "@/service/network/PageSpeedProvider.js"; +import { HardwareProvider } from "@/service/network/HardwareProvider.js"; +import { DockerProvider } from "@/service/network/DockerProvider.js"; +import { PortProvider } from "@/service/network/PortProvider.js"; +import { GameProvider } from "@/service/network/GameProvider.js"; +import { GrpcProvider } from "@/service/network/GrpcProvider.js"; +import { WebSocketProvider } from "@/service/network/WebSocketProvider.js"; +import { DNSProvider } from "@/service/network/DNSProvider.js"; +export interface WorkerServices { + worker: IQueueWorker; + networkService: INetworkService; + bufferService: IBufferService; + statusService: IStatusService; +} + +export const buildWorker = async (shared: SharedServices, envSettings: EnvConfig): Promise => { + const { + logger, + settingsService, + checkService, + geoChecksService, + notificationsService, + incidentService, + workerId, + jobsRepository, + queueWorkersRepository, + monitorsRepository, + checksRepository, + geoChecksRepository, + monitorStatsRepository, + incidentsRepository, + teamsRepository, + maintenanceWindowsRepository, + } = shared; + + // *********************** + // Network providers + // *********************** + const pingProvider = new PingProvider(ping); + const httpProvider = new HttpProvider(got, new AdvancedMatcher(jmespath)); + const pageSpeedProvider = new PageSpeedProvider(httpProvider, settingsService, logger); + const hardwareProvider = new HardwareProvider(httpProvider); + const dockerProvider = new DockerProvider(logger, Docker); + const portProvider = new PortProvider(net); + const gameProvider = new GameProvider(logger, GameDig); + const grpcProvider = new GrpcProvider(grpc, protoLoader); + const webSocketProvider = new WebSocketProvider(WebSocket); + const dnsProvider = new DNSProvider(() => new Resolver()); + + const networkService = new NetworkService(axios, logger, [ + pingProvider, + httpProvider, + pageSpeedProvider, + hardwareProvider, + dockerProvider, + portProvider, + gameProvider, + grpcProvider, + webSocketProvider, + dnsProvider, + ]); + + const bufferService = new BufferService(logger, checkService, geoChecksService, settingsService, jobsRepository); + const statusService = new StatusService(logger, monitorsRepository, monitorStatsRepository); + const monitorStatusPolicy = new MonitorStatusPolicy(); + + // *********************** + // Reactors and dispatcher + // Handles notifications and incidents + // *********************** + + const notificationReactor = new NotificationReactor(notificationsService); + const incidentReactor = new IncidentReactor(incidentService); + const reactorDispatcher = new ReactorDispatcher(logger, [notificationReactor, incidentReactor]); + + // *********************** + // Check producer/evaluator + // Handles creating and evaluatiog checks + // *********************** + const checkProducer = new CheckProducer(monitorsRepository, maintenanceWindowsRepository, checkService, networkService, bufferService, logger); + const checkEvaluator = new CheckEvaluator(statusService, monitorStatusPolicy); + const geoCheckPipeline = new GeoChecksPipeline(maintenanceWindowsRepository, geoChecksService, bufferService, logger); + + // *********************** + // Worker + // *********************** + + const workerHelper = new WorkerHelper( + logger, + checkService, + settingsService, + monitorsRepository, + jobsRepository, + teamsRepository, + monitorStatsRepository, + checksRepository, + incidentsRepository, + geoChecksRepository + ); + + const worker = await DBQueueWorker.create( + logger, + jobsRepository, + monitorsRepository, + checksRepository, + checkService, + checkProducer, + checkEvaluator, + geoCheckPipeline, + reactorDispatcher, + workerHelper, + queueWorkersRepository, + envSettings.queueMode, + envSettings.queuePrimaryProcesses, + workerId + ); + + return { worker, networkService, bufferService, statusService }; +}; diff --git a/server/src/domain/checks/check.model.ts b/server/src/domain/checks/check.model.ts index 82be6937c..c3fbc23ff 100644 --- a/server/src/domain/checks/check.model.ts +++ b/server/src/domain/checks/check.model.ts @@ -241,6 +241,7 @@ const CheckSchema = new Schema( }, errors: { type: [errorSchema], + suppressReservedKeysWarning: true, default: () => [], }, capture: { diff --git a/server/src/index.ts b/server/src/index.ts index 37d4e59d8..d121d7b25 100755 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,4 +1,7 @@ -import { initializeServices } from "./config/services.js"; +import { buildShared } from "@/config/services.shared.js"; +import { buildApi } from "@/config/services.api.js"; +import { buildWorker } from "@/config/services.worker.js"; + import { initializeControllers } from "./config/controllers.js"; import { createApp } from "./app.js"; import { initShutdownListener } from "@/shutdown.js"; @@ -9,44 +12,83 @@ import fs from "fs"; import Logger, { ILogger } from "@/utils/logger.js"; import { SettingsService } from "@/domain/app-settings/app-settings.service.js"; - +import { JobScheduler } from "@/worker/worker.job-scheduler.js"; +import { IJobScheduler } from "@/worker/worker.interface.js"; const SERVICE_NAME = "Server"; let logger: ILogger; const startApp = async () => { - // Validate environment variables first - const env = validateEnv(); + // *********************** + // Basic setup + // *********************** - // Create settings service (env only — DB repository injected after connect) + const env = validateEnv(); const settingsService = new SettingsService(env); const envSettings = settingsService.loadSettings(); - // Create logger logger = new Logger({ envSettings }); + const { queueMode, queuePrimaryProcesses } = envSettings; - // Initialize services (connects DB, creates repositories, injects settingsRepository) - const services = await initializeServices({ logger, envSettings, settingsService }); + logger.info({ + message: `Queue mode: ${queueMode}`, + service: SERVICE_NAME, + method: "startApp", + }); - // If this is a worker instance, we're done. No need for express - if (envSettings.queueMode === "worker") { - logger.info({ - message: "Worker instance started. API will not be started", - service: SERVICE_NAME, - }); - initShutdownListener(null, services); + logger.info({ + message: `Process: ${queuePrimaryProcesses}`, + service: SERVICE_NAME, + method: "startApp", + }); + + // *********************** + // Build shared services + // *********************** + const shared = await buildShared({ logger, envSettings, settingsService }); + + // *********************** + // Worker node path, don't need API + // *********************** + if (queueMode === "worker") { + const { worker } = await buildWorker(shared, envSettings); + logger.info({ message: "Worker instance started. API will not be started", service: SERVICE_NAME }); + initShutdownListener(null, { worker, db: shared.db }); return; } - // FE path + // *********************** + // Primary node path + // *********************** + let scheduler: IJobScheduler; + + // *********************** + //Primary node processes jobs, need full worker + // *********************** + if (queuePrimaryProcesses === true) { + const { worker } = await buildWorker(shared, envSettings); + scheduler = worker; + } + // *********************** + // Primary node does not process jobs, only need scheduler + // *********************** + else { + scheduler = new JobScheduler(shared.jobsRepository, shared.queueWorkersRepository, shared.workerId); + } + + const services = buildApi(shared, scheduler); + + // *********************** + // FE assets always needed for primary node + // *********************** + const __filename = fileURLToPath(import.meta.url); const __dirname = path.dirname(__filename); const openApiSpec = JSON.parse(fs.readFileSync(path.join(__dirname, "../openapi.json"), "utf8")); const frontendPath = path.join(__dirname, "..", "public"); - // Initialize controllers const controllers = initializeControllers(services); const app = createApp({ - services, + apiServices: services, controllers, envSettings, frontendPath, diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index 38d1d5114..584cfea39 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -1,8 +1,14 @@ -import { InitializedServices } from "./config/services.js"; +import { IDb } from "@/db/db.interface.js"; import { logger } from "./utils/logger.js"; import type { Server } from "http"; +import { Mongoose } from "mongoose"; +import { IJobScheduler } from "@/worker/worker.interface.js"; -export const initShutdownListener = (server: Server | null, services: InitializedServices) => { +type ShutdownTargets = { + worker: IJobScheduler; + db: IDb; +}; +export const initShutdownListener = (server: Server | null, services: ShutdownTargets) => { const SERVICE_NAME = "Server"; let isShuttingDown = false; From 74783fd7381d594869f9918f7e57ca3820beac52 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 24 Jun 2026 10:53:03 -0700 Subject: [PATCH 5/6] clean up logger --- server/src/api/middleware/handleErrors.ts | 4 +-- server/src/app.ts | 2 +- server/src/config/envValidation.ts | 28 +++++++++++++++---- server/src/db/db.mongo.ts | 2 +- .../0003_cleanupDuplicateMonitorStats.ts | 4 +-- .../0004_fixInfrastructureThresholds.ts | 4 +-- .../0005_migrateStatusPageTypeToArray.ts | 4 +-- ...anupDuplicateMonitorStatsForUniqueIndex.ts | 4 +-- ...igrateMaintenanceWindowMonitorIdToArray.ts | 4 +-- .../0008_backfillMonitorLastEvaluatedAt.ts | 4 +-- server/src/db/migration/index.ts | 16 +++++------ server/src/domain/monitors/monitor.model.ts | 2 +- server/src/index.ts | 7 +++-- server/src/shutdown.ts | 4 ++- server/src/utils/logger.ts | 19 +++++-------- 15 files changed, 61 insertions(+), 47 deletions(-) diff --git a/server/src/api/middleware/handleErrors.ts b/server/src/api/middleware/handleErrors.ts index 239349b90..77817f12e 100755 --- a/server/src/api/middleware/handleErrors.ts +++ b/server/src/api/middleware/handleErrors.ts @@ -1,12 +1,12 @@ import type { NextFunction, Request, Response } from "express"; import { MulterError } from "multer"; -import { logger } from "@/utils/logger.js"; +import type { ILogger } from "@/utils/logger.js"; import { AppError } from "@/utils/AppError.js"; // Multer throws MulterError, send 413 Payload Too Large / 400 const multerErrorStatus = (error: MulterError): number => (error.code === "LIMIT_FILE_SIZE" ? 413 : 400); -const handleErrors = (error: unknown, req: Request, res: Response, _next: NextFunction) => { +const handleErrors = (logger: ILogger) => (error: unknown, req: Request, res: Response, _next: NextFunction) => { const status = error instanceof MulterError ? multerErrorStatus(error) : error instanceof AppError ? error.status || 500 : 500; const message = error instanceof AppError ? error.message : error instanceof Error ? error.message : "Server error"; const service = error instanceof AppError ? error.service : error instanceof MulterError ? "uploadMiddleware" : "unknownService"; diff --git a/server/src/app.ts b/server/src/app.ts index d559edb1f..94a122137 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -112,6 +112,6 @@ export const createApp = ({ app.get("*", (req, res) => { res.sendFile(path.join(frontendPath, "index.html")); }); - app.use(handleErrors); + app.use(handleErrors(services.logger)); return app; }; diff --git a/server/src/config/envValidation.ts b/server/src/config/envValidation.ts index b65ef5165..bdfacf906 100644 --- a/server/src/config/envValidation.ts +++ b/server/src/config/envValidation.ts @@ -1,6 +1,7 @@ import { z } from "zod"; import { DbTypes, QueueModes } from "@/domain/app-settings/app-settings.type.js"; import { booleanCoercion } from "@/api/validation/shared.js"; +import { ILogger } from "@/utils/logger.js"; const envSchema = z.object({ // Server Configuration @@ -31,24 +32,39 @@ const envSchema = z.object({ export type ValidatedEnv = z.infer; -export const validateEnv = (): ValidatedEnv => { +export const validateEnv = (logger: ILogger): ValidatedEnv => { const result = envSchema.safeParse(process.env); if (!result.success) { - console.error("Environment validation failed:"); - console.error(""); + logger.error({ + message: "Env validation failed", + method: "validateEnv", + service: "Server", + }); const errors = result.error.format(); for (const [key, value] of Object.entries(errors)) { if (key !== "_errors" && value && typeof value === "object" && "_errors" in value) { - console.error(` ${key}: ${value._errors.join(", ")}`); + logger.error({ + message: `${key}: ${value._errors.join(", ")}`, + method: "validateEnv", + service: "Server", + }); } } - console.error("Please check your .env file and ensure all required variables are set."); + logger.error({ + message: "Please check your .env file and ensure all required variables are set.", + method: "validateEnv", + service: "Server", + }); process.exit(1); } - console.log("Environment variables validated successfully"); + logger.info({ + message: "Environment variables validated successfully", + method: "validateEnv", + service: "Server", + }); return result.data; }; diff --git a/server/src/db/db.mongo.ts b/server/src/db/db.mongo.ts index 2585eb58b..5aca5550a 100755 --- a/server/src/db/db.mongo.ts +++ b/server/src/db/db.mongo.ts @@ -42,7 +42,7 @@ class MongoDB implements IDb { // the unique index on MonitorStats.monitorId is enforced, otherwise databases // with stray duplicates (caused by the pre-fix race in updateRunningStats) // will crash at startup when the unique index fails to build. - await runMigrations(); + await runMigrations(this.logger); // Sync indexes const models = mongoose.modelNames(); diff --git a/server/src/db/migration/0003_cleanupDuplicateMonitorStats.ts b/server/src/db/migration/0003_cleanupDuplicateMonitorStats.ts index f772b5229..39a67d91e 100644 --- a/server/src/db/migration/0003_cleanupDuplicateMonitorStats.ts +++ b/server/src/db/migration/0003_cleanupDuplicateMonitorStats.ts @@ -1,12 +1,12 @@ import { MonitorStatsModel } from "../../domain/monitor-stats/monitor-stats.model.js"; -import { logger } from "@/utils/logger.js"; +import type { ILogger } from "@/utils/logger.js"; /** * Cleanup duplicate MonitorStats documents * Keeps the most recent document (by updatedAt) for each monitorId * and deletes all older duplicates */ -export async function cleanupDuplicateMonitorStats(): Promise { +export async function cleanupDuplicateMonitorStats(logger: ILogger): Promise { const SERVICE_NAME = "Migration:CleanupDuplicateMonitorStats"; try { diff --git a/server/src/db/migration/0004_fixInfrastructureThresholds.ts b/server/src/db/migration/0004_fixInfrastructureThresholds.ts index dfc45c4fe..c21d552a6 100644 --- a/server/src/db/migration/0004_fixInfrastructureThresholds.ts +++ b/server/src/db/migration/0004_fixInfrastructureThresholds.ts @@ -1,11 +1,11 @@ import { MonitorModel } from "../../domain/monitors/monitor.model.js"; -import { logger } from "@/utils/logger.js"; +import type { ILogger } from "@/utils/logger.js"; /** * Fix infrastructure monitors that had thresholds incorrectly set to 0 or 5 * due to the old default value. Updates them to 100 (disabled). */ -export async function fixInfrastructureThresholds(): Promise { +export async function fixInfrastructureThresholds(logger: ILogger): Promise { const SERVICE_NAME = "Migration:FixInfrastructureThresholds"; try { diff --git a/server/src/db/migration/0005_migrateStatusPageTypeToArray.ts b/server/src/db/migration/0005_migrateStatusPageTypeToArray.ts index d5dcafbd1..d75b069e0 100644 --- a/server/src/db/migration/0005_migrateStatusPageTypeToArray.ts +++ b/server/src/db/migration/0005_migrateStatusPageTypeToArray.ts @@ -1,8 +1,8 @@ import mongoose from "mongoose"; -import { logger } from "@/utils/logger.js"; +import type { ILogger } from "@/utils/logger.js"; import StatusPageModel from "../../domain/status-pages/status-page.model.js"; -export async function migrateStatusPageTypeToArray(): Promise { +export async function migrateStatusPageTypeToArray(logger: ILogger): Promise { const SERVICE_NAME = "Migration:MigrateStatusPageTypeToArray"; try { diff --git a/server/src/db/migration/0006_cleanupDuplicateMonitorStatsForUniqueIndex.ts b/server/src/db/migration/0006_cleanupDuplicateMonitorStatsForUniqueIndex.ts index 4a34e9926..2479f83c6 100644 --- a/server/src/db/migration/0006_cleanupDuplicateMonitorStatsForUniqueIndex.ts +++ b/server/src/db/migration/0006_cleanupDuplicateMonitorStatsForUniqueIndex.ts @@ -1,5 +1,5 @@ import { MonitorStatsModel } from "../../domain/monitor-stats/monitor-stats.model.js"; -import { logger } from "@/utils/logger.js"; +import type { ILogger } from "@/utils/logger.js"; /** * Second-pass dedupe of MonitorStats before the unique index on monitorId is enforced. @@ -11,7 +11,7 @@ import { logger } from "@/utils/logger.js"; * Keeps the document with the highest totalChecks for each monitorId (most data) and * deletes the rest. Falls back to most recent updatedAt when totalChecks tie. */ -export async function cleanupDuplicateMonitorStatsForUniqueIndex(): Promise { +export async function cleanupDuplicateMonitorStatsForUniqueIndex(logger: ILogger): Promise { const SERVICE_NAME = "Migration:CleanupDuplicateMonitorStatsForUniqueIndex"; try { diff --git a/server/src/db/migration/0007_migrateMaintenanceWindowMonitorIdToArray.ts b/server/src/db/migration/0007_migrateMaintenanceWindowMonitorIdToArray.ts index bf5766000..437c66704 100644 --- a/server/src/db/migration/0007_migrateMaintenanceWindowMonitorIdToArray.ts +++ b/server/src/db/migration/0007_migrateMaintenanceWindowMonitorIdToArray.ts @@ -1,6 +1,6 @@ import mongoose from "mongoose"; import type { AnyBulkWriteOperation, Document } from "mongodb"; -import { logger } from "@/utils/logger.js"; +import type { ILogger } from "@/utils/logger.js"; import MaintenanceWindowModel from "../../domain/maintenance-windows/maintenance-window.model.js"; type GroupResult = { @@ -18,7 +18,7 @@ type GroupResult = { monitorIds: mongoose.Types.ObjectId[]; }; -export async function migrateMaintenanceWindowMonitorIdToArray(): Promise { +export async function migrateMaintenanceWindowMonitorIdToArray(logger: ILogger): Promise { const SERVICE_NAME = "Migration:MigrateMaintenanceWindowMonitorIdToArray"; try { diff --git a/server/src/db/migration/0008_backfillMonitorLastEvaluatedAt.ts b/server/src/db/migration/0008_backfillMonitorLastEvaluatedAt.ts index deb7cb02e..148c4b74d 100644 --- a/server/src/db/migration/0008_backfillMonitorLastEvaluatedAt.ts +++ b/server/src/db/migration/0008_backfillMonitorLastEvaluatedAt.ts @@ -1,6 +1,6 @@ import mongoose from "mongoose"; import { MonitorModel } from "../../domain/monitors/monitor.model.js"; -import { logger } from "@/utils/logger.js"; +import type { ILogger } from "@/utils/logger.js"; const JOBS_COLLECTION = "jobs"; @@ -9,7 +9,7 @@ const JOBS_COLLECTION = "jobs"; * * Stamp the upgrade moment, not 0. If set to 0, all checks will be replayed */ -export async function backfillMonitorLastEvaluatedAt(): Promise { +export async function backfillMonitorLastEvaluatedAt(logger: ILogger): Promise { const SERVICE_NAME = "Migration:BackfillMonitorLastEvaluatedAt"; try { diff --git a/server/src/db/migration/index.ts b/server/src/db/migration/index.ts index d68a2468e..0b0745d19 100644 --- a/server/src/db/migration/index.ts +++ b/server/src/db/migration/index.ts @@ -11,7 +11,7 @@ import type { ILogger } from "@/utils/logger.js"; type MigrationEntry = { name: string; - execute: () => Promise; + execute: (logger: ILogger) => Promise; }; const migrations: MigrationEntry[] = [ @@ -25,34 +25,34 @@ const migrations: MigrationEntry[] = [ { name: "0008_backfillMonitorLastEvaluatedAt", execute: backfillMonitorLastEvaluatedAt }, ]; -const runMigrations = async (logger?: ILogger) => { +const runMigrations = async (logger: ILogger) => { try { - logger?.info({ message: "Running migrations", service: "Migrations" }); + logger.info({ message: "Running migrations", service: "Migrations" }); for (const migration of migrations) { const exists = await MigrationModel.findOne({ name: migration.name, status: "completed" }); if (exists) { - logger?.info({ message: `Skipping ${migration.name}`, service: "Migrations" }); + logger.info({ message: `Skipping ${migration.name}`, service: "Migrations" }); continue; } try { - await migration.execute(); + await migration.execute(logger); await MigrationModel.findOneAndUpdate( { name: migration.name }, { status: "completed", completedAt: new Date(), error: undefined }, { upsert: true } ); - logger?.info({ message: `Completed ${migration.name}`, service: "Migrations" }); + logger.info({ message: `Completed ${migration.name}`, service: "Migrations" }); } catch (error) { const err = error as Error; await MigrationModel.findOneAndUpdate({ name: migration.name }, { status: "failed", error: err?.message }, { upsert: true }); throw error; } } - logger?.info({ message: "Migrations completed", service: "Migrations" }); + logger.info({ message: "Migrations completed", service: "Migrations" }); } catch (error) { const err = error as Error; - logger?.error({ message: "Migration failed", service: "Migrations", details: { error: err?.message }, stack: err?.stack }); + logger.error({ message: "Migration failed", service: "Migrations", details: { error: err?.message }, stack: err?.stack }); throw error; } }; diff --git a/server/src/domain/monitors/monitor.model.ts b/server/src/domain/monitors/monitor.model.ts index 8a095658c..47aa70b5e 100644 --- a/server/src/domain/monitors/monitor.model.ts +++ b/server/src/domain/monitors/monitor.model.ts @@ -197,7 +197,7 @@ const checkSnapshotSchema = new Schema( audits: { type: snapshotAuditsSchema }, createdAt: { type: Date, required: true }, }, - { _id: false } + { _id: false, suppressReservedKeysWarning: true } ); const MonitorSchema = new Schema( diff --git a/server/src/index.ts b/server/src/index.ts index d121d7b25..3b0ab6ba4 100755 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -22,10 +22,11 @@ const startApp = async () => { // Basic setup // *********************** - const env = validateEnv(); + logger = new Logger(); + const env = validateEnv(logger); + logger.setLogLevel(env.LOG_LEVEL); const settingsService = new SettingsService(env); const envSettings = settingsService.loadSettings(); - logger = new Logger({ envSettings }); const { queueMode, queuePrimaryProcesses } = envSettings; logger.info({ @@ -51,7 +52,7 @@ const startApp = async () => { if (queueMode === "worker") { const { worker } = await buildWorker(shared, envSettings); logger.info({ message: "Worker instance started. API will not be started", service: SERVICE_NAME }); - initShutdownListener(null, { worker, db: shared.db }); + initShutdownListener(null, { worker, db: shared.db, logger }); return; } diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index 584cfea39..625547e85 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -1,5 +1,5 @@ import { IDb } from "@/db/db.interface.js"; -import { logger } from "./utils/logger.js"; +import type { ILogger } from "@/utils/logger.js"; import type { Server } from "http"; import { Mongoose } from "mongoose"; import { IJobScheduler } from "@/worker/worker.interface.js"; @@ -7,9 +7,11 @@ import { IJobScheduler } from "@/worker/worker.interface.js"; type ShutdownTargets = { worker: IJobScheduler; db: IDb; + logger: ILogger; }; export const initShutdownListener = (server: Server | null, services: ShutdownTargets) => { const SERVICE_NAME = "Server"; + const logger = services.logger; let isShuttingDown = false; diff --git a/server/src/utils/logger.ts b/server/src/utils/logger.ts index c53e7112c..41142fd92 100755 --- a/server/src/utils/logger.ts +++ b/server/src/utils/logger.ts @@ -1,7 +1,6 @@ import { createLogger, format, transports, Logger as WinstonLogger } from "winston"; import type { Logform } from "winston"; import dotenv from "dotenv"; -import { EnvConfig } from "@/domain/app-settings/app-settings.service.js"; dotenv.config(); const SERVICE_NAME = "Logger"; @@ -28,18 +27,17 @@ export interface ILogger { cacheLog(entry: LogEntry): void; getLogs(): LogEntry[]; buildLogEntry(level: string, config: LogConfig): LogEntry; + setLogLevel(level: string): void; } class Logger implements ILogger { static SERVICE_NAME = SERVICE_NAME; private logger: WinstonLogger; - private envSettings: Partial; private logCache: LogEntry[]; private maxCacheSize: number; - constructor({ envSettings }: { envSettings: Partial }) { - this.envSettings = envSettings; + constructor() { this.logCache = []; this.maxCacheSize = 1000; const consoleFormat = format.printf((info: Logform.TransformableInfo) => { @@ -84,10 +82,8 @@ class Logger implements ILogger { return msg; }); - const logLevel = this.envSettings.logLevel || "info"; - this.logger = createLogger({ - level: logLevel, + level: "info", format: format.combine(format.timestamp()), transports: [ new transports.Console({ @@ -151,11 +147,10 @@ class Logger implements ILogger { timestamp: new Date().toISOString(), }; } + + setLogLevel(level: string): void { + this.logger.level = level || "info"; + } } export default Logger; - -// Legacy logger -const logger = new Logger({ envSettings: { logLevel: "debug" } }); -export { logger }; -export type { LogConfig, LogEntry }; From 0f230c4d70e20c56e41c7edc2c17f514ab290ffe Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 24 Jun 2026 10:53:13 -0700 Subject: [PATCH 6/6] update tests --- .../test/integration/backfillLastEvaluatedAt.test.ts | 11 +++++++---- server/test/unit/middleware/handleErrors.test.ts | 11 +++++++---- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/server/test/integration/backfillLastEvaluatedAt.test.ts b/server/test/integration/backfillLastEvaluatedAt.test.ts index ec0459b7d..f4c8a58c5 100644 --- a/server/test/integration/backfillLastEvaluatedAt.test.ts +++ b/server/test/integration/backfillLastEvaluatedAt.test.ts @@ -3,6 +3,9 @@ import mongoose from "mongoose"; import { MongoMemoryServer } from "mongodb-memory-server"; import { MonitorModel } from "../../src/domain/monitors/monitor.model.ts"; import { backfillMonitorLastEvaluatedAt } from "../../src/db/migration/0008_backfillMonitorLastEvaluatedAt.ts"; +import { createMockLogger } from "../helpers/createMockLogger.ts"; + +const logger = createMockLogger(); // ── Real-Mongo harness ───────────────────────────────────────────────────────── // Legacy monitors are inserted through the raw driver, NOT MonitorModel.create — the @@ -30,7 +33,7 @@ describe("0008_backfillMonitorLastEvaluatedAt", () => { await MonitorModel.collection.insertOne({ name: "legacy" }); // pre-upgrade doc, no lastEvaluatedAt const before = Date.now(); - await backfillMonitorLastEvaluatedAt(); + await backfillMonitorLastEvaluatedAt(logger); const doc = await MonitorModel.collection.findOne({ name: "legacy" }); // Must be a recent timestamp — backfilling 0 would replay the whole check history on the first evaluate. @@ -42,7 +45,7 @@ describe("0008_backfillMonitorLastEvaluatedAt", () => { it("leaves monitors that already have lastEvaluatedAt untouched (idempotent)", async () => { await MonitorModel.collection.insertOne({ name: "current", lastEvaluatedAt: 12345 }); - await backfillMonitorLastEvaluatedAt(); + await backfillMonitorLastEvaluatedAt(logger); const doc = await MonitorModel.collection.findOne({ name: "current" }); expect(doc?.lastEvaluatedAt).toBe(12345); @@ -52,13 +55,13 @@ describe("0008_backfillMonitorLastEvaluatedAt", () => { const db = mongoose.connection.db!; await db.collection("jobs").insertOne({ id: "legacy-job", template: "monitor-job" }); // old-scheduler row - await backfillMonitorLastEvaluatedAt(); + await backfillMonitorLastEvaluatedAt(logger); const exists = await db.listCollections({ name: "jobs" }).toArray(); expect(exists).toHaveLength(0); }); it("does not throw when there is no jobs collection to drop", async () => { - await expect(backfillMonitorLastEvaluatedAt()).resolves.toBeUndefined(); + await expect(backfillMonitorLastEvaluatedAt(logger)).resolves.toBeUndefined(); }); }); diff --git a/server/test/unit/middleware/handleErrors.test.ts b/server/test/unit/middleware/handleErrors.test.ts index 5994cf7d3..3ce49adbb 100644 --- a/server/test/unit/middleware/handleErrors.test.ts +++ b/server/test/unit/middleware/handleErrors.test.ts @@ -3,6 +3,9 @@ import type { NextFunction, Request, Response } from "express"; import { MulterError } from "multer"; import { handleErrors } from "../../../src/api/middleware/handleErrors.ts"; import { AppError } from "../../../src/utils/AppError.ts"; +import { createMockLogger } from "../../helpers/createMockLogger.ts"; + +const logger = createMockLogger(); const makeRes = (): Response => { const res = {} as Response; @@ -14,25 +17,25 @@ const makeRes = (): Response => { describe("handleErrors", () => { it("maps a multer file-size violation to 413 Payload Too Large", () => { const res = makeRes(); - handleErrors(new MulterError("LIMIT_FILE_SIZE", "profileImage"), {} as Request, res, (() => {}) as NextFunction); + handleErrors(logger)(new MulterError("LIMIT_FILE_SIZE", "profileImage"), {} as Request, res, (() => {}) as NextFunction); expect(res.status).toHaveBeenCalledWith(413); }); it("maps other multer errors to 400", () => { const res = makeRes(); - handleErrors(new MulterError("LIMIT_UNEXPECTED_FILE", "profileImage"), {} as Request, res, (() => {}) as NextFunction); + handleErrors(logger)(new MulterError("LIMIT_UNEXPECTED_FILE", "profileImage"), {} as Request, res, (() => {}) as NextFunction); expect(res.status).toHaveBeenCalledWith(400); }); it("preserves AppError status codes", () => { const res = makeRes(); - handleErrors(new AppError({ status: 415, message: "bad image" }), {} as Request, res, (() => {}) as NextFunction); + handleErrors(logger)(new AppError({ status: 415, message: "bad image" }), {} as Request, res, (() => {}) as NextFunction); expect(res.status).toHaveBeenCalledWith(415); }); it("falls back to 500 for unknown errors", () => { const res = makeRes(); - handleErrors(new Error("boom"), {} as Request, res, (() => {}) as NextFunction); + handleErrors(logger)(new Error("boom"), {} as Request, res, (() => {}) as NextFunction); expect(res.status).toHaveBeenCalledWith(500); }); });