From ee9c08b1bb6caef55a55df1e717793991bf01c60 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 10 Jun 2026 13:12:51 -0700 Subject: [PATCH] extract reactors and pipeline --- server/src/config/services.ts | 43 ++- .../src/worker/reactors/reactor.dispatcher.ts | 39 +++ .../src/worker/reactors/reactor.incident.ts | 13 + .../src/worker/reactors/reactor.interface.ts | 7 + .../worker/reactors/reactor.notification.ts | 13 + server/src/worker/worker.check-pipeline.ts | 187 +++++++++++++ server/src/worker/worker.helper.ts | 255 +++--------------- server/src/worker/worker.interface.ts | 11 + .../worker/worker.monitor-status-policy.ts | 46 ++++ 9 files changed, 380 insertions(+), 234 deletions(-) create mode 100644 server/src/worker/reactors/reactor.dispatcher.ts create mode 100644 server/src/worker/reactors/reactor.incident.ts create mode 100644 server/src/worker/reactors/reactor.interface.ts create mode 100644 server/src/worker/reactors/reactor.notification.ts create mode 100644 server/src/worker/worker.check-pipeline.ts create mode 100644 server/src/worker/worker.monitor-status-policy.ts diff --git a/server/src/config/services.ts b/server/src/config/services.ts index dcd6cbb98..8d872201e 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -28,12 +28,13 @@ import { IUserService, UserService } from "@/domain/users/user.service.js"; import { BufferService, IBufferService } from "@/service/bufferService.js"; import { EmailService, IEmailService } from "@/service/emailService.js"; import { GlobalPingService } from "@/service/globalPingService.js"; -import { QueueHelper } from "@/worker/worker.helper.js"; +import { WorkerHelper } from "@/worker/worker.helper.js"; import { IWorker } from "@/worker/worker.interface.js"; import { LessSimpleWorker } from "@/worker/worker.less-simple.js"; import { SuperSimpleQueue } from "@/worker/worker.super-simple.js"; import { INetworkService, NetworkService } from "@/service/networkService.js"; import { IStatusService, StatusService } from "@/service/statusService.js"; +import { MonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js"; // Network providers import { PingProvider } from "@/service/network/PingProvider.js"; @@ -102,6 +103,10 @@ import MongoUsersRepository from "@/domain/users/user.repository.mongo.js"; import { ILogger } from "@/utils/logger.js"; import { AppError } from "@/utils/AppError.js"; import { type QueueMode } from "@/domain/app-settings/app-settings.type.js"; +import { NotificationReactor } from "@/worker/reactors/reactor.notification.js"; +import { ReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js"; +import { IncidentReactor } from "@/worker/reactors/reactor.incident.js"; +import { CheckPipeline, GeoChecksPipeline } from "@/worker/worker.check-pipeline.js"; export type InitializedServices = { settingsService: ISettingsService; @@ -309,23 +314,39 @@ export const initializeServices = async ({ const tagsService = new TagsService(tagsRepository, monitorsRepository); - const queueHelper = new QueueHelper( - logger, + const monitorStatusPolicy = new MonitorStatusPolicy(); + // reactors and dispatcher + const notificationReactor = new NotificationReactor(notificationsService); + const incidentReactor = new IncidentReactor(incidentService); + const reactorDispatcher = new ReactorDispatcher(logger, [notificationReactor, incidentReactor]); + + // pipelines + const checkPipeline = new CheckPipeline( + monitorsRepository, + maintenanceWindowsRepository, + checkService, networkService, + bufferService, + monitorStatusPolicy, statusService, - notificationsService, + logger + ); + + const geoCheckPipeline = new GeoChecksPipeline(maintenanceWindowsRepository, geoChecksService, bufferService, logger); + + const workerHelper = new WorkerHelper( + logger, checkService, settingsService, - bufferService, - incidentService, - maintenanceWindowsRepository, monitorsRepository, teamsRepository, monitorStatsRepository, checksRepository, incidentsRepository, - geoChecksService, - geoChecksRepository + geoChecksRepository, + reactorDispatcher, + checkPipeline, + geoCheckPipeline ); if (queueMode === "worker" && envSettings.queueType !== "lessSimpleQueue") { @@ -340,10 +361,10 @@ export const initializeServices = async ({ let worker: IWorker; switch (envSettings.queueType) { case "lessSimpleQueue": - worker = await LessSimpleWorker.create(logger, queueHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode); + worker = await LessSimpleWorker.create(logger, workerHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode); break; default: - worker = await SuperSimpleQueue.create(logger, queueHelper, monitorsRepository); + worker = await SuperSimpleQueue.create(logger, workerHelper, monitorsRepository); } // Business services diff --git a/server/src/worker/reactors/reactor.dispatcher.ts b/server/src/worker/reactors/reactor.dispatcher.ts new file mode 100644 index 000000000..a4ffe757e --- /dev/null +++ b/server/src/worker/reactors/reactor.dispatcher.ts @@ -0,0 +1,39 @@ +import { ILogger } from "@/utils/logger.js"; +import { MonitorEvaluation } from "@/worker/worker.interface.js"; +import { IMonitorReactor } from "@/worker/reactors/reactor.interface.js"; + +const SERVICE_NAME = "ReactorDispatcher"; + +export interface IReactorDispatcher { + dispatch(evaluation: MonitorEvaluation): Promise; +} + +export class ReactorDispatcher implements IReactorDispatcher { + constructor( + private logger: ILogger, + private reactors: IMonitorReactor[] + ) {} + + private logFailure(reactor: IMonitorReactor, evaluation: MonitorEvaluation, error: unknown) { + this.logger.error({ + service: SERVICE_NAME, + method: "dispatch", + message: `Reactor ${reactor.name} failed for monitor ${evaluation.monitor.id}: ${error instanceof Error ? error.message : String(error)}`, + stack: error instanceof Error ? error.stack : undefined, + }); + } + + dispatch = async (evaluation: MonitorEvaluation) => { + for (const reactor of this.reactors) { + if (reactor.blocking) { + try { + await reactor.react(evaluation); + } catch (error: unknown) { + this.logFailure(reactor, evaluation, error); + } + } else { + reactor.react(evaluation).catch((error: unknown) => this.logFailure(reactor, evaluation, error)); + } + } + }; +} diff --git a/server/src/worker/reactors/reactor.incident.ts b/server/src/worker/reactors/reactor.incident.ts new file mode 100644 index 000000000..3679a30cf --- /dev/null +++ b/server/src/worker/reactors/reactor.incident.ts @@ -0,0 +1,13 @@ +import { IMonitorReactor } from "@/worker/reactors/reactor.interface.js"; +import { MonitorEvaluation } from "@/worker/worker.interface.js"; +import { IIncidentService } from "@/domain/incidents/incident.service.js"; + +export class IncidentReactor implements IMonitorReactor { + readonly name = "incident"; + readonly blocking = true; + constructor(private incidentService: IIncidentService) {} + + react = async (evaluation: MonitorEvaluation) => { + await this.incidentService.handleIncident(evaluation.monitor, evaluation.statusChange.code, evaluation.decision, evaluation.status); + }; +} diff --git a/server/src/worker/reactors/reactor.interface.ts b/server/src/worker/reactors/reactor.interface.ts new file mode 100644 index 000000000..f9ed6e436 --- /dev/null +++ b/server/src/worker/reactors/reactor.interface.ts @@ -0,0 +1,7 @@ +import { MonitorEvaluation } from "@/worker/worker.interface.js"; + +export interface IMonitorReactor { + readonly name: string; + readonly blocking: boolean; + react(evaluation: MonitorEvaluation): Promise; +} diff --git a/server/src/worker/reactors/reactor.notification.ts b/server/src/worker/reactors/reactor.notification.ts new file mode 100644 index 000000000..1daf528f4 --- /dev/null +++ b/server/src/worker/reactors/reactor.notification.ts @@ -0,0 +1,13 @@ +import { INotificationsService } from "@/domain/notifications/notification.service.js"; +import { IMonitorReactor } from "@/worker/reactors/reactor.interface.js"; +import { MonitorEvaluation } from "@/worker/worker.interface.js"; +export class NotificationReactor implements IMonitorReactor { + readonly name = "notification"; + readonly blocking = false; + constructor(private notificationService: INotificationsService) {} + + react = async (evaluation: MonitorEvaluation) => { + if (!evaluation.decision.shouldSendNotification) return; + await this.notificationService.handleNotifications(evaluation.monitor, evaluation.status, evaluation.decision); + }; +} diff --git a/server/src/worker/worker.check-pipeline.ts b/server/src/worker/worker.check-pipeline.ts new file mode 100644 index 000000000..f0958fa20 --- /dev/null +++ b/server/src/worker/worker.check-pipeline.ts @@ -0,0 +1,187 @@ +import { Monitor } from "@/domain/monitors/monitor.types.js"; +import { MonitorEvaluation } from "@/worker/worker.interface.js"; +import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js"; +import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; +import { isWindowActive } from "@/utils/maintenanceWindow.js"; +import { ILogger } from "@/utils/logger.js"; +import { INetworkService } from "@/service/networkService.js"; +import { ICheckService } from "@/domain/checks/check.service.js"; +import { IBufferService } from "@/service/bufferService.js"; +import { IMonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js"; +import { IStatusService } from "@/service/statusService.js"; +import { AppError } from "@/utils/AppError.js"; +import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; +import { IGeoChecksService } from "@/domain/geo-checks/geo-check.service.js"; + +const SERVICE_NAME = "CheckPipeline"; + +export interface ICheckPipeline { + run(monitor: Monitor): Promise; // null = skipped +} + +export class CheckPipeline implements ICheckPipeline { + constructor( + private monitorsRepository: IMonitorsRepository, + private maintenanceWindowsRepository: IMaintenanceWindowsRepository, + private checkService: ICheckService, + private networkService: INetworkService, + private bufferService: IBufferService, + private monitorStatusPolicy: IMonitorStatusPolicy, + private statusService: IStatusService, + private logger: ILogger + ) {} + + private async isInMaintenanceWindow(monitorId: string, teamId: string) { + const maintenanceWindows = await this.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId); + const now = new Date(); + return maintenanceWindows.some((window) => isWindowActive(window, now)); + } + + run = async (monitor: Monitor): Promise => { + const monitorId = monitor.id; + if (!monitorId) { + throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getMonitorJob" }); + } + // **************************** + // Step 1: Acquire + // **************************** + + // Step 1a: Maintenance window gate - skip if in maintenance + const maintenanceWindowActive = await this.isInMaintenanceWindow(monitor.id, monitor.teamId); + if (maintenanceWindowActive) { + this.logger.debug({ + message: `Monitor ${monitor.id} is in maintenance window`, + service: SERVICE_NAME, + method: "runCheckPipeline", + }); + if (monitor.status !== "maintenance") { + await this.monitorsRepository.updateById(monitor.id, monitor.teamId, { status: "maintenance" }); + } + return null; + } + + // Step 1b: Acquire status + const status = await this.networkService.requestStatus(monitor); + if (!status) { + throw new Error("No network response"); + } + + // **************************** + // Step 2: Record + // **************************** + + // Step 2a: Create & record a check, return null if fail + const check = this.checkService.buildCheck(status); + if (!check) { + this.logger.warn({ + message: `No check could be built for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "runCHeckPipeline", + details: { code: status.code, message: status.message }, + }); + return null; + } + // Step 2b: Add to buffer + this.bufferService.addToBuffer(check); + + // **************************** + // Step 3: Evaluate and return result to reactors + // **************************** + const statusChangeResult = await this.statusService.updateMonitorStatus(status, check); + + // Step 5. Get decisions and create an evaluation obj + const decision = this.monitorStatusPolicy.evaluate(statusChangeResult); + const evaluation: MonitorEvaluation = { + monitor: statusChangeResult.monitor, + status, + check, + statusChange: statusChangeResult, + decision, + }; + return evaluation; + }; +} + +export class GeoChecksPipeline implements ICheckPipeline { + constructor( + private maintenanceWindowsRepository: IMaintenanceWindowsRepository, + private geoChecksService: IGeoChecksService, + private bufferService: IBufferService, + private logger: ILogger + ) {} + + private async isInMaintenanceWindow(monitorId: string, teamId: string) { + const maintenanceWindows = await this.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId); + const now = new Date(); + return maintenanceWindows.some((window) => isWindowActive(window, now)); + } + + run = async (monitor: Monitor): Promise => { + // **************************** + // Step 1: Acquire + // **************************** + + // Step 1a: Guards - skip if unsupported or not enabled + + if (!monitor.id) { + throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getHeartbeatGeoJob" }); + } + + if (!monitor.geoCheckEnabled) { + return null; + } + if (!supportsGeoCheck(monitor.type)) { + this.logger.debug({ + message: `Monitor ${monitor.id} type does not support geo checks, skipping`, + service: SERVICE_NAME, + method: "runGeoChecksPipeline", + }); + return null; + } + + if (!monitor.geoCheckLocations || monitor.geoCheckLocations.length === 0) { + this.logger.warn({ + message: `No geo check locations configured for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "runGeoChecksPipeline", + }); + return null; + } + + // Step 1b: Maintenance window check + const maintenanceWindowActive = await this.isInMaintenanceWindow(monitor.id, monitor.teamId); + if (maintenanceWindowActive) { + this.logger.debug({ + message: `Monitor ${monitor.id} is in maintenance window, skipping geo check`, + service: SERVICE_NAME, + method: "runGeoChecksPipeline", + }); + return null; + } + + // **************************** + // Step 2: Record + // **************************** + + // Step 2a: Build geo check, return null if fail + const geoCheck = await this.geoChecksService.buildGeoCheck(monitor); + if (!geoCheck) { + this.logger.warn({ + message: `No geo check could be built for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "runGeoChecksPipeline", + }); + return null; + } + + // Step 2b: Add to buffer + this.bufferService.addGeoCheckToBuffer(geoCheck); + + this.logger.debug({ + message: `Geo check job executed for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "runGeoChecksPipeline", + }); + return null; + }; +} diff --git a/server/src/worker/worker.helper.ts b/server/src/worker/worker.helper.ts index 6840ff695..29de9f2a3 100644 --- a/server/src/worker/worker.helper.ts +++ b/server/src/worker/worker.helper.ts @@ -1,26 +1,17 @@ const SERVICE_NAME = "JobQueueHelper"; import type { Monitor } from "@/domain/monitors/monitor.types.js"; -import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; -import { AppError } from "@/utils/AppError.js"; import { ISettingsService } from "@/domain/app-settings/app-settings.service.js"; import { ICheckService } from "@/domain/checks/check.service.js"; -import { type IGeoChecksService } from "@/domain/geo-checks/geo-check.service.js"; -import { IncidentService } from "@/domain/incidents/incident.service.js"; -import { INotificationsService } from "@/domain/notifications/notification.service.js"; -import { INetworkService } from "@/service/networkService.js"; -import { IStatusService } from "@/service/statusService.js"; import { CHECK_TTL_SENTINEL } from "@/domain/checks/check.type.js"; -import { type StatusChangeResult } from "@/types/network.js"; -import { isWindowActive } from "@/utils/maintenanceWindow.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 { 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 { ITeamsRepository } from "@/domain/teams/team.repository.interface.js"; import { ILogger } from "@/utils/logger.js"; -import { IBufferService } from "@/service/bufferService.js"; +import { IReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js"; +import { ICheckPipeline } from "@/worker/worker.check-pipeline.js"; export interface IWorkerHelper { readonly serviceName: string; @@ -28,7 +19,6 @@ export interface IWorkerHelper { getHeartbeatGeoJob(): (monitor: Monitor) => Promise; getCleanupOrphanedJob(): () => Promise; getCleanupRetentionJob(): () => Promise; - isInMaintenanceWindow(monitorId: string, teamId: string): Promise; } export interface MonitorActionDecision { @@ -45,138 +35,59 @@ export interface MonitorActionDecision { }; } -export class QueueHelper implements IWorkerHelper { +export class WorkerHelper implements IWorkerHelper { static SERVICE_NAME = SERVICE_NAME; private logger: ILogger; - private networkService: INetworkService; - private statusService: IStatusService; - private notificationsService: INotificationsService; private checkService: ICheckService; private settingsService: ISettingsService; - private buffer: IBufferService; - private incidentService: IncidentService; - private maintenanceWindowsRepository: IMaintenanceWindowsRepository; private monitorsRepository: IMonitorsRepository; private teamsRepository: ITeamsRepository; private monitorStatsRepository: IMonitorStatsRepository; private checksRepository: IChecksRepository; private incidentsRepository: IIncidentsRepository; - private geoChecksService: IGeoChecksService; private geoChecksRepository: IGeoChecksRepository; + private reactorDispatcher: IReactorDispatcher; + private checkPipeline: ICheckPipeline; + private geoCheckPipeline: ICheckPipeline; constructor( logger: ILogger, - networkService: INetworkService, - statusService: IStatusService, - notificationsService: INotificationsService, checkService: ICheckService, settingsService: ISettingsService, - buffer: IBufferService, - incidentService: IncidentService, - maintenanceWindowsRepository: IMaintenanceWindowsRepository, monitorsRepository: IMonitorsRepository, teamsRepository: ITeamsRepository, monitorStatsRepository: IMonitorStatsRepository, checksRepository: IChecksRepository, incidentsRepository: IIncidentsRepository, - geoChecksService: IGeoChecksService, - geoChecksRepository: IGeoChecksRepository + geoChecksRepository: IGeoChecksRepository, + reactorDispatcher: IReactorDispatcher, + checkPipeline: ICheckPipeline, + geoCheckPipeline: ICheckPipeline ) { this.logger = logger; - this.networkService = networkService; - this.statusService = statusService; this.checkService = checkService; this.settingsService = settingsService; - this.buffer = buffer; - this.notificationsService = notificationsService; - this.incidentService = incidentService; - this.maintenanceWindowsRepository = maintenanceWindowsRepository; this.monitorsRepository = monitorsRepository; this.teamsRepository = teamsRepository; this.monitorStatsRepository = monitorStatsRepository; this.checksRepository = checksRepository; this.incidentsRepository = incidentsRepository; - this.geoChecksService = geoChecksService; this.geoChecksRepository = geoChecksRepository; + this.reactorDispatcher = reactorDispatcher; + this.checkPipeline = checkPipeline; + this.geoCheckPipeline = geoCheckPipeline; } get serviceName() { - return QueueHelper.SERVICE_NAME; + return WorkerHelper.SERVICE_NAME; } getHeartbeatJob = () => { return async (monitor: Monitor) => { try { - const monitorId = monitor.id; - const teamId = monitor.teamId; - if (!monitorId) { - throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getMonitorJob" }); - } - - // Step 1. Check for maintenance window, if found, skip the check - - const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId, teamId); - if (maintenanceWindowActive) { - this.logger.debug({ - message: `Monitor ${monitorId} is in maintenance window`, - service: SERVICE_NAME, - method: "getMonitorJob", - }); - if (monitor.status !== "maintenance") { - await this.monitorsRepository.updateById(monitorId, teamId, { status: "maintenance" }); - } - return; - } - - // Step 2. Request monitor status - const status = await this.networkService.requestStatus(monitor); - if (!status) { - throw new Error("No network response"); - } - - // Step 3. Build check - const check = this.checkService.buildCheck(status); - if (!check) { - this.logger.warn({ - message: `No check could be built for monitor ${monitorId}`, - service: SERVICE_NAME, - method: "getMonitorJob", - details: { code: status.code, message: status.message }, - }); - return; - } - // Step 4 Add check to buffer - this.buffer.addToBuffer(check); - // Step 4. Update monitor status - const statusChangeResult = await this.statusService.updateMonitorStatus(status, check); - - // Step 5. Get decisions - const decision = this.evaluateMonitorAction(statusChangeResult); - - // Step 6. Handle notifications (best effort, continue even in event of failure, don't wait) - if (decision.shouldSendNotification) { - this.notificationsService.handleNotifications(statusChangeResult.monitor, status, decision).catch((error: unknown) => { - this.logger.error({ - message: `Error sending notifications for job ${statusChangeResult.monitor.id}: ${error instanceof Error ? error.message : "Unknown error"}`, - service: SERVICE_NAME, - method: "getMonitorJob", - stack: error instanceof Error ? error.stack : undefined, - }); - }); - } - - // Step 7. Handle incidents - try { - await this.incidentService.handleIncident(statusChangeResult.monitor, statusChangeResult.code, decision, status); - } catch (error: unknown) { - this.logger.warn({ - message: `Error handling incident for job ${monitor.id}: ${error instanceof Error ? error.message : "Unknown error"}`, - service: SERVICE_NAME, - method: "getMonitorJob", - stack: error instanceof Error ? error.stack : undefined, - }); - } + const evaluation = await this.checkPipeline.run(monitor); + if (evaluation) await this.reactorDispatcher.dispatch(evaluation); } catch (error: unknown) { this.logger.warn({ message: error instanceof Error ? error.message : "Unknown error", @@ -189,6 +100,22 @@ export class QueueHelper implements IWorkerHelper { }; }; + getHeartbeatGeoJob = () => { + return async (monitor: Monitor) => { + try { + await this.geoCheckPipeline.run(monitor); + } catch (error: unknown) { + this.logger.error({ + message: error instanceof Error ? error.message : "Unknown error", + service: SERVICE_NAME, + method: "getHeartbeatGeoJob", + stack: error instanceof Error ? error.stack : undefined, + }); + // Don't throw - geo check failures shouldn't crash the job scheduler + } + }; + }; + getCleanupOrphanedJob = () => { return async () => { try { @@ -280,86 +207,6 @@ export class QueueHelper implements IWorkerHelper { }; }; - getHeartbeatGeoJob = () => { - return async (monitor: Monitor) => { - try { - const monitorId = monitor.id; - const teamId = monitor.teamId; - - // Step 1: Validate monitor eligibility - if (!monitorId) { - throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getHeartbeatGeoJob" }); - } - - if (!monitor.geoCheckEnabled) { - return; - } - if (!supportsGeoCheck(monitor.type)) { - this.logger.debug({ - message: `Monitor ${monitorId} type does not support geo checks, skipping`, - service: SERVICE_NAME, - method: "getHeartbeatGeoJob", - }); - return; - } - - if (!monitor.geoCheckLocations || monitor.geoCheckLocations.length === 0) { - this.logger.warn({ - message: `No geo check locations configured for monitor ${monitorId}`, - service: SERVICE_NAME, - method: "getHeartbeatGeoJob", - }); - return; - } - - // Step 2: Check for maintenance window - const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId, teamId); - if (maintenanceWindowActive) { - this.logger.debug({ - message: `Monitor ${monitorId} is in maintenance window, skipping geo check`, - service: SERVICE_NAME, - method: "getHeartbeatGeoJob", - }); - return; - } - - // Step 3: Build geo check (handles API calls and polling) - const geoCheck = await this.geoChecksService.buildGeoCheck(monitor); - if (!geoCheck) { - this.logger.warn({ - message: `No geo check could be built for monitor ${monitorId}`, - service: SERVICE_NAME, - method: "getHeartbeatGeoJob", - }); - return; - } - - // Step 4: Add geo check to buffer - this.buffer.addGeoCheckToBuffer(geoCheck); - - this.logger.debug({ - message: `Geo check job executed for monitor ${monitorId}`, - service: SERVICE_NAME, - method: "getHeartbeatGeoJob", - }); - } catch (error: unknown) { - this.logger.error({ - message: error instanceof Error ? error.message : "Unknown error", - service: SERVICE_NAME, - method: "getHeartbeatGeoJob", - stack: error instanceof Error ? error.stack : undefined, - }); - // Don't throw - geo check failures shouldn't crash the job scheduler - } - }; - }; - - async isInMaintenanceWindow(monitorId: string, teamId: string) { - const maintenanceWindows = await this.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId); - const now = new Date(); - return maintenanceWindows.some((window) => isWindowActive(window, now)); - } - getCleanupRetentionJob = () => { return async () => { try { @@ -393,42 +240,4 @@ export class QueueHelper implements IWorkerHelper { } }; }; - - private evaluateMonitorAction(statusChangeResult: StatusChangeResult): MonitorActionDecision { - const { monitor, statusChanged, prevStatus } = statusChangeResult; - - // Initialize result - const decision: MonitorActionDecision = { - shouldCreateIncident: false, - shouldResolveIncident: false, - shouldSendNotification: false, - incidentReason: null, - notificationReason: null, - }; - - if (!statusChanged) { - return decision; - } - - if (monitor.status === "down") { - // Monitor went down (unreachable) - decision.shouldCreateIncident = true; - decision.shouldSendNotification = true; - decision.incidentReason = "status_down"; - decision.notificationReason = "status_change"; - } else if (monitor.status === "breached") { - // Hardware monitor exceeded thresholds - decision.shouldCreateIncident = true; - decision.shouldSendNotification = true; - decision.incidentReason = "threshold_breach"; - decision.notificationReason = "threshold_breach"; - } else if (monitor.status === "up" && (prevStatus === "down" || prevStatus === "breached")) { - // Monitor recovered from down or breached state - decision.shouldResolveIncident = true; - decision.shouldSendNotification = true; - decision.notificationReason = "status_change"; - } - - return decision; - } } diff --git a/server/src/worker/worker.interface.ts b/server/src/worker/worker.interface.ts index 45dc94ca1..5d1ccf97f 100644 --- a/server/src/worker/worker.interface.ts +++ b/server/src/worker/worker.interface.ts @@ -1,5 +1,16 @@ import { Monitor } from "@/domain/monitors/monitor.types.js"; import { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js"; +import { Check } from "@/domain/checks/check.type.js"; +import { MonitorPayloadMap, MonitorStatusResponse, StatusChangeResult } from "@/types/network.js"; +import { MonitorActionDecision } from "@/worker/worker.helper.js"; + +export type MonitorEvaluation = { + monitor: Monitor; + status: MonitorStatusResponse; // raw result from networkService.requestStatus + check: Check; + statusChange: StatusChangeResult; // from statusService.updateMonitorStatus + decision: MonitorActionDecision; // from MonitorStatusPolicy +}; export type WorkerJobFailure = { monitorId: string | number; diff --git a/server/src/worker/worker.monitor-status-policy.ts b/server/src/worker/worker.monitor-status-policy.ts new file mode 100644 index 000000000..4ef36b2a9 --- /dev/null +++ b/server/src/worker/worker.monitor-status-policy.ts @@ -0,0 +1,46 @@ +import { StatusChangeResult } from "@/types/network.js"; +import { MonitorActionDecision } from "@/worker/worker.helper.js"; + +export interface IMonitorStatusPolicy { + evaluate(statusChange: StatusChangeResult): MonitorActionDecision; +} + +export class MonitorStatusPolicy implements IMonitorStatusPolicy { + evaluate = (statusChangeResult: StatusChangeResult): MonitorActionDecision => { + const { monitor, statusChanged, prevStatus } = statusChangeResult; + + // Initialize result + const decision: MonitorActionDecision = { + shouldCreateIncident: false, + shouldResolveIncident: false, + shouldSendNotification: false, + incidentReason: null, + notificationReason: null, + }; + + if (!statusChanged) { + return decision; + } + + if (monitor.status === "down") { + // Monitor went down (unreachable) + decision.shouldCreateIncident = true; + decision.shouldSendNotification = true; + decision.incidentReason = "status_down"; + decision.notificationReason = "status_change"; + } else if (monitor.status === "breached") { + // Hardware monitor exceeded thresholds + decision.shouldCreateIncident = true; + decision.shouldSendNotification = true; + decision.incidentReason = "threshold_breach"; + decision.notificationReason = "threshold_breach"; + } else if (monitor.status === "up" && (prevStatus === "down" || prevStatus === "breached")) { + // Monitor recovered from down or breached state + decision.shouldResolveIncident = true; + decision.shouldSendNotification = true; + decision.notificationReason = "status_change"; + } + + return decision; + }; +}