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; + }; +} diff --git a/server/test/helpers/heartbeatTestHarness.ts b/server/test/helpers/heartbeatTestHarness.ts index 79dae92cd..ea78d5a5d 100644 --- a/server/test/helpers/heartbeatTestHarness.ts +++ b/server/test/helpers/heartbeatTestHarness.ts @@ -1,5 +1,10 @@ import { jest } from "@jest/globals"; -import { QueueHelper } from "../../src/worker/worker.helper.ts"; +import { WorkerHelper } from "../../src/worker/worker.helper.ts"; +import { MonitorStatusPolicy } from "../../src/worker/worker.monitor-status-policy.ts"; +import { CheckPipeline, GeoChecksPipeline } from "../../src/worker/worker.check-pipeline.ts"; +import { NotificationReactor } from "../../src/worker/reactors/reactor.notification.ts"; +import { IncidentReactor } from "../../src/worker/reactors/reactor.incident.ts"; +import { ReactorDispatcher } from "../../src/worker/reactors/reactor.dispatcher.ts"; import { StatusService } from "../../src/service/statusService.ts"; import { IncidentService } from "../../src/domain/incidents/incident.service.ts"; import { InMemoryMonitorsRepository } from "./InMemoryMonitorsRepository.ts"; @@ -142,24 +147,37 @@ export function createHeartbeatTestHarness(): HeartbeatTestHarness { }; const maintenanceWindowsRepo = { findByMonitorId: jest.fn().mockResolvedValue([]) }; + const geoChecksService = { buildGeoCheck: jest.fn() }; - const helper = new QueueHelper( - logger, + const checkPipeline = new CheckPipeline( + monitorsRepo as any, + maintenanceWindowsRepo as any, + checkService as any, networkService as any, + bufferStub as any, + new MonitorStatusPolicy(), statusService as any, - notificationsService as any, + logger + ); + const geoCheckPipeline = new GeoChecksPipeline(maintenanceWindowsRepo as any, geoChecksService as any, bufferStub as any, logger); + + const notificationReactor = new NotificationReactor(notificationsService as any); + const incidentReactor = new IncidentReactor(incidentService as any); + const reactorDispatcher = new ReactorDispatcher(logger, [notificationReactor, incidentReactor]); + + const helper = new WorkerHelper( + logger, checkService as any, { getDBSettings: jest.fn() } as any, - bufferStub as any, - incidentService as any, - maintenanceWindowsRepo as any, monitorsRepo as any, { findAllTeamIds: jest.fn() } as any, createStubMonitorStatsRepo() as any, createStubChecksRepo() as any, incidentsRepo as any, - { buildGeoCheck: jest.fn() } as any, - { deleteByMonitorIdsNotIn: jest.fn() } as any + { deleteByMonitorIdsNotIn: jest.fn() } as any, + reactorDispatcher, + checkPipeline, + geoCheckPipeline ); return { diff --git a/server/test/unit/services/checkPipeline.test.ts b/server/test/unit/services/checkPipeline.test.ts new file mode 100644 index 000000000..20ab65950 --- /dev/null +++ b/server/test/unit/services/checkPipeline.test.ts @@ -0,0 +1,217 @@ +import { describe, expect, it, jest } from "@jest/globals"; +import { CheckPipeline, GeoChecksPipeline } from "../../../src/worker/worker.check-pipeline.ts"; +import { MonitorStatusPolicy } from "../../../src/worker/worker.monitor-status-policy.ts"; +import type { Monitor } from "../../../src/domain/monitors/monitor.types.ts"; +import { createMockLogger } from "../../helpers/createMockLogger.ts"; + +const makeMonitor = (overrides?: Partial): Monitor => + ({ + id: "m1", + teamId: "team", + type: "http", + interval: 60000, + status: "up", + geoCheckEnabled: false, + geoCheckLocations: [], + ...overrides, + }) as Monitor; + +const activeWindow = () => { + const now = Date.now(); + return { active: true, start: new Date(now - 1000).toISOString(), end: new Date(now + 1000).toISOString(), repeat: 0 }; +}; + +// ── CheckPipeline ──────────────────────────────────────────────────────────── + +const createCheckPipeline = (overrides?: Record) => { + const defaults = { + logger: createMockLogger(), + networkService: { requestStatus: jest.fn().mockResolvedValue({ monitorId: "m1", status: true, code: 200, message: "OK" }) }, + statusService: { + updateMonitorStatus: jest.fn().mockResolvedValue({ monitor: { id: "m1", status: "up" }, statusChanged: false, prevStatus: "up", code: 200 }), + }, + checkService: { buildCheck: jest.fn().mockReturnValue({ id: "check-1" }) }, + buffer: { addToBuffer: jest.fn() }, + monitorsRepository: { updateById: jest.fn().mockResolvedValue({}) }, + maintenanceWindowsRepository: { findByMonitorId: jest.fn().mockResolvedValue([]) }, + monitorStatusPolicy: new MonitorStatusPolicy(), + ...overrides, + }; + const pipeline = new CheckPipeline( + defaults.monitorsRepository as any, + defaults.maintenanceWindowsRepository as any, + defaults.checkService as any, + defaults.networkService as any, + defaults.buffer as any, + defaults.monitorStatusPolicy as any, + defaults.statusService as any, + defaults.logger as any + ); + return { pipeline, defaults }; +}; + +describe("CheckPipeline", () => { + it("throws when monitor id is missing", async () => { + const { pipeline } = createCheckPipeline(); + await expect(pipeline.run({} as Monitor)).rejects.toThrow("No monitor id"); + }); + + it("skips and sets maintenance status when in maintenance window", async () => { + const { pipeline, defaults } = createCheckPipeline({ + maintenanceWindowsRepository: { findByMonitorId: jest.fn().mockResolvedValue([activeWindow()]) }, + }); + + const result = await pipeline.run(makeMonitor({ status: "up" })); + + expect(result).toBeNull(); + expect(defaults.networkService.requestStatus).not.toHaveBeenCalled(); + expect(defaults.monitorsRepository.updateById).toHaveBeenCalledWith("m1", "team", { status: "maintenance" }); + }); + + it("does not re-write status when already in maintenance", async () => { + const { pipeline, defaults } = createCheckPipeline({ + maintenanceWindowsRepository: { findByMonitorId: jest.fn().mockResolvedValue([activeWindow()]) }, + }); + + await pipeline.run(makeMonitor({ status: "maintenance" })); + + expect(defaults.monitorsRepository.updateById).not.toHaveBeenCalled(); + }); + + it("throws when network response is null", async () => { + const { pipeline } = createCheckPipeline({ + networkService: { requestStatus: jest.fn().mockResolvedValue(null) }, + }); + await expect(pipeline.run(makeMonitor())).rejects.toThrow("No network response"); + }); + + it("returns null and warns when buildCheck returns null", async () => { + const { pipeline, defaults } = createCheckPipeline({ + checkService: { buildCheck: jest.fn().mockReturnValue(null) }, + }); + + const result = await pipeline.run(makeMonitor()); + + expect(result).toBeNull(); + expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("No check could be built") })); + expect(defaults.statusService.updateMonitorStatus).not.toHaveBeenCalled(); + }); + + it("buffers the check, updates status, and returns an evaluation", async () => { + const status = { monitorId: "m1", status: false, code: 500, message: "Error" }; + const statusChange = { monitor: { id: "m1", status: "down" }, statusChanged: true, prevStatus: "up", code: 500 }; + const { pipeline, defaults } = createCheckPipeline({ + networkService: { requestStatus: jest.fn().mockResolvedValue(status) }, + statusService: { updateMonitorStatus: jest.fn().mockResolvedValue(statusChange) }, + }); + + const result = await pipeline.run(makeMonitor()); + + expect(defaults.buffer.addToBuffer).toHaveBeenCalledWith({ id: "check-1" }); + expect(defaults.statusService.updateMonitorStatus).toHaveBeenCalled(); + expect(result).toMatchObject({ + monitor: statusChange.monitor, + status, + check: { id: "check-1" }, + statusChange, + decision: expect.objectContaining({ shouldCreateIncident: true, shouldSendNotification: true, incidentReason: "status_down" }), + }); + }); + + it("proceeds with the check when no maintenance window is active", async () => { + const { pipeline, defaults } = createCheckPipeline({ + maintenanceWindowsRepository: { findByMonitorId: jest.fn().mockResolvedValue([{ ...activeWindow(), active: false }]) }, + }); + + await pipeline.run(makeMonitor()); + + expect(defaults.networkService.requestStatus).toHaveBeenCalled(); + }); +}); + +// ── GeoChecksPipeline ────────────────────────────────────────────────────────── + +const createGeoPipeline = (overrides?: Record) => { + const defaults = { + logger: createMockLogger(), + geoChecksService: { buildGeoCheck: jest.fn().mockResolvedValue(null) }, + buffer: { addGeoCheckToBuffer: jest.fn() }, + maintenanceWindowsRepository: { findByMonitorId: jest.fn().mockResolvedValue([]) }, + ...overrides, + }; + const pipeline = new GeoChecksPipeline( + defaults.maintenanceWindowsRepository as any, + defaults.geoChecksService as any, + defaults.buffer as any, + defaults.logger as any + ); + return { pipeline, defaults }; +}; + +const geoMonitor = (overrides?: Partial) => + makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckLocations: ["us-east"], ...overrides }); + +describe("GeoChecksPipeline", () => { + it("throws when monitor id is missing", async () => { + const { pipeline } = createGeoPipeline(); + await expect(pipeline.run({} as Monitor)).rejects.toThrow("No monitor id"); + }); + + it("returns null when geoCheckEnabled is false", async () => { + const { pipeline, defaults } = createGeoPipeline(); + const result = await pipeline.run(makeMonitor({ geoCheckEnabled: false })); + expect(result).toBeNull(); + expect(defaults.geoChecksService.buildGeoCheck).not.toHaveBeenCalled(); + }); + + it("returns null when monitor type does not support geo checks", async () => { + const { pipeline, defaults } = createGeoPipeline(); + await pipeline.run(geoMonitor({ type: "hardware" })); + expect(defaults.logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("does not support geo checks") })); + expect(defaults.geoChecksService.buildGeoCheck).not.toHaveBeenCalled(); + }); + + it("warns when geoCheckLocations is empty", async () => { + const { pipeline, defaults } = createGeoPipeline(); + await pipeline.run(geoMonitor({ geoCheckLocations: [] })); + expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("No geo check locations") })); + }); + + it("warns when geoCheckLocations is undefined", async () => { + const { pipeline, defaults } = createGeoPipeline(); + await pipeline.run(geoMonitor({ geoCheckLocations: undefined as any })); + expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("No geo check locations") })); + }); + + it("skips when in maintenance window", async () => { + const { pipeline, defaults } = createGeoPipeline({ + maintenanceWindowsRepository: { findByMonitorId: jest.fn().mockResolvedValue([activeWindow()]) }, + }); + await pipeline.run(geoMonitor()); + expect(defaults.logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("maintenance window") })); + expect(defaults.geoChecksService.buildGeoCheck).not.toHaveBeenCalled(); + }); + + it("warns when buildGeoCheck returns null", async () => { + const { pipeline, defaults } = createGeoPipeline(); + await pipeline.run(geoMonitor()); + expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("No geo check could be built") })); + }); + + it("adds geo check to buffer on success", async () => { + const geoCheck = { id: "gc-1", monitorId: "m1" }; + const { pipeline, defaults } = createGeoPipeline({ + geoChecksService: { buildGeoCheck: jest.fn().mockResolvedValue(geoCheck) }, + }); + await pipeline.run(geoMonitor()); + expect(defaults.buffer.addGeoCheckToBuffer).toHaveBeenCalledWith(geoCheck); + expect(defaults.logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("Geo check job executed") })); + }); + + it("propagates errors (the job glue handles isolation)", async () => { + const { pipeline } = createGeoPipeline({ + geoChecksService: { buildGeoCheck: jest.fn().mockRejectedValue(new Error("api timeout")) }, + }); + await expect(pipeline.run(geoMonitor())).rejects.toThrow("api timeout"); + }); +}); diff --git a/server/test/unit/services/monitorStatusPolicy.test.ts b/server/test/unit/services/monitorStatusPolicy.test.ts new file mode 100644 index 000000000..7b340a3e9 --- /dev/null +++ b/server/test/unit/services/monitorStatusPolicy.test.ts @@ -0,0 +1,74 @@ +import { describe, expect, it } from "@jest/globals"; +import { MonitorStatusPolicy } from "../../../src/worker/worker.monitor-status-policy.ts"; +import type { StatusChangeResult } from "../../../src/types/network.ts"; + +// ── Helpers ────────────────────────────────────────────────────────────────── + +const makeStatusChange = (overrides: Partial & { status?: string }): StatusChangeResult => { + const { status = "up", ...rest } = overrides; + return { + monitor: { id: "m1", status } as any, + statusChanged: false, + prevStatus: "up", + code: 200, + timestamp: 0, + ...rest, + } as StatusChangeResult; +}; + +// ── Tests ──────────────────────────────────────────────────────────────────── + +describe("MonitorStatusPolicy", () => { + const policy = new MonitorStatusPolicy(); + + it("does nothing when statusChanged is false", () => { + const decision = policy.evaluate(makeStatusChange({ status: "down", statusChanged: false, prevStatus: "up" })); + expect(decision).toEqual({ + shouldCreateIncident: false, + shouldResolveIncident: false, + shouldSendNotification: false, + incidentReason: null, + notificationReason: null, + }); + }); + + it("creates incident and notifies when monitor goes down", () => { + const decision = policy.evaluate(makeStatusChange({ status: "down", statusChanged: true, prevStatus: "up", code: 500 })); + expect(decision).toMatchObject({ + shouldCreateIncident: true, + shouldSendNotification: true, + incidentReason: "status_down", + notificationReason: "status_change", + }); + }); + + it("creates incident and notifies when monitor is breached", () => { + const decision = policy.evaluate(makeStatusChange({ status: "breached", statusChanged: true, prevStatus: "up" })); + expect(decision).toMatchObject({ + shouldCreateIncident: true, + shouldSendNotification: true, + incidentReason: "threshold_breach", + notificationReason: "threshold_breach", + }); + }); + + it("resolves incident when monitor recovers from down", () => { + const decision = policy.evaluate(makeStatusChange({ status: "up", statusChanged: true, prevStatus: "down" })); + expect(decision).toMatchObject({ + shouldResolveIncident: true, + shouldSendNotification: true, + shouldCreateIncident: false, + notificationReason: "status_change", + }); + }); + + it("resolves incident when monitor recovers from breached", () => { + const decision = policy.evaluate(makeStatusChange({ status: "up", statusChanged: true, prevStatus: "breached" })); + expect(decision).toMatchObject({ shouldResolveIncident: true, shouldSendNotification: true }); + }); + + it("does not create or resolve for unhandled status transitions", () => { + const decision = policy.evaluate(makeStatusChange({ status: "paused", statusChanged: true, prevStatus: "up" })); + expect(decision).toMatchObject({ shouldCreateIncident: false, shouldResolveIncident: false, shouldSendNotification: false }); + }); +}); diff --git a/server/test/unit/services/reactorDispatcher.test.ts b/server/test/unit/services/reactorDispatcher.test.ts new file mode 100644 index 000000000..0bcc90526 --- /dev/null +++ b/server/test/unit/services/reactorDispatcher.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it, jest } from "@jest/globals"; +import { ReactorDispatcher } from "../../../src/worker/reactors/reactor.dispatcher.ts"; +import { createMockLogger } from "../../helpers/createMockLogger.ts"; + +const evaluation = { monitor: { id: "m1" } } as any; + +const makeReactor = (name: string, blocking: boolean, impl?: () => Promise) => ({ + name, + blocking, + react: jest.fn(impl ?? (() => Promise.resolve())), +}); + +describe("ReactorDispatcher", () => { + it("runs every reactor with the evaluation", async () => { + const a = makeReactor("a", true); + const b = makeReactor("b", false); + const dispatcher = new ReactorDispatcher(createMockLogger() as any, [a, b] as any); + + await dispatcher.dispatch(evaluation); + + expect(a.react).toHaveBeenCalledWith(evaluation); + expect(b.react).toHaveBeenCalledWith(evaluation); + }); + + it("awaits blocking reactors before resolving", async () => { + let done = false; + const blocking = makeReactor( + "incident", + true, + () => + new Promise((r) => + setTimeout(() => { + done = true; + r(); + }, 10) + ) + ); + const dispatcher = new ReactorDispatcher(createMockLogger() as any, [blocking] as any); + + await dispatcher.dispatch(evaluation); + + expect(done).toBe(true); + }); + + it("does not await non-blocking reactors", async () => { + let done = false; + const nonBlocking = makeReactor( + "notification", + false, + () => + new Promise((r) => + setTimeout(() => { + done = true; + r(); + }, 20) + ) + ); + const dispatcher = new ReactorDispatcher(createMockLogger() as any, [nonBlocking] as any); + + await dispatcher.dispatch(evaluation); + + expect(done).toBe(false); // resolved before the slow fire-and-forget reactor finished + }); + + it("isolates a throwing blocking reactor and still runs later reactors", async () => { + const logger = createMockLogger(); + const bad = makeReactor("incident", true, () => Promise.reject(new Error("db error"))); + const good = makeReactor("after", true); + const dispatcher = new ReactorDispatcher(logger as any, [bad, good] as any); + + await expect(dispatcher.dispatch(evaluation)).resolves.toBeUndefined(); + + expect(good.react).toHaveBeenCalled(); + expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("db error") })); + }); + + it("isolates a rejecting non-blocking reactor (fire-and-forget)", async () => { + const logger = createMockLogger(); + const bad = makeReactor("notification", false, () => Promise.reject(new Error("smtp down"))); + const dispatcher = new ReactorDispatcher(logger as any, [bad] as any); + + await dispatcher.dispatch(evaluation); + await new Promise((r) => setTimeout(r, 10)); // let the detached catch run + + expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("smtp down") })); + }); +}); diff --git a/server/test/unit/services/superSimpleQueueHelper.test.ts b/server/test/unit/services/superSimpleQueueHelper.test.ts index bb4a7c001..1c6f28861 100644 --- a/server/test/unit/services/superSimpleQueueHelper.test.ts +++ b/server/test/unit/services/superSimpleQueueHelper.test.ts @@ -1,14 +1,17 @@ import { describe, expect, it, jest } from "@jest/globals"; -import { QueueHelper } from "../../../src/worker/worker.helper.ts"; +import { WorkerHelper } from "../../../src/worker/worker.helper.ts"; import type { Monitor } from "../../../src/domain/monitors/monitor.types.ts"; import { createMockLogger } from "../../helpers/createMockLogger.ts"; // ── Helpers ────────────────────────────────────────────────────────────────── +// +// WorkerHelper is now thin glue: it wires the scheduler's job functions to the +// CheckPipeline / GeoChecksPipeline (acquire→record→decide) and the +// ReactorDispatcher (react), and owns the cleanup jobs. The pipeline and +// dispatcher internals are tested in checkPipeline.test.ts / reactorDispatcher.test.ts; +// here we test only the glue + cleanup jobs. const createHelper = (overrides?: Record) => { - const maintenanceWindowsRepository = { - findByMonitorId: jest.fn().mockResolvedValue([]), - }; const monitorsRepository = { updateById: jest.fn().mockResolvedValue({}), findAllMonitorIds: jest.fn().mockResolvedValue(["m1"]), @@ -29,55 +32,48 @@ const createHelper = (overrides?: Record) => { const geoChecksRepository = { deleteByMonitorIdsNotIn: jest.fn().mockResolvedValue(0), }; - const statusServiceMock = { - updateMonitorStatus: jest.fn().mockResolvedValue({ monitor: { id: "m1", status: "up" }, statusChanged: false, prevStatus: "up", code: 200 }), - }; - const settingsServiceMock = { + const settingsService = { getDBSettings: jest.fn().mockResolvedValue({ checkTTL: 30 }), }; - const geoChecksServiceMock = { - buildGeoCheck: jest.fn().mockResolvedValue(null), + const checkService = { + deleteOlderThan: jest.fn().mockResolvedValue(0), }; + const checkPipeline = { run: jest.fn().mockResolvedValue(null) }; + const geoCheckPipeline = { run: jest.fn().mockResolvedValue(null) }; + const reactorDispatcher = { dispatch: jest.fn().mockResolvedValue(undefined) }; + const defaults = { logger: createMockLogger(), - networkService: { requestStatus: jest.fn() }, - statusService: statusServiceMock, - notificationsService: { handleNotifications: jest.fn().mockResolvedValue(undefined) }, - checkService: { buildCheck: jest.fn().mockReturnValue({ id: "check-1" }), deleteOlderThan: jest.fn().mockResolvedValue(0) }, - settingsService: settingsServiceMock, - buffer: { addToBuffer: jest.fn(), addGeoCheckToBuffer: jest.fn() }, - incidentService: { handleIncident: jest.fn().mockResolvedValue(undefined) }, - maintenanceWindowsRepository, + checkService, + settingsService, monitorsRepository, teamsRepository, monitorStatsRepository, checksRepository, incidentsRepository, - geoChecksService: geoChecksServiceMock, geoChecksRepository, + reactorDispatcher, + checkPipeline, + geoCheckPipeline, ...overrides, }; - const helper = new QueueHelper( + const helper = new WorkerHelper( defaults.logger as any, - defaults.networkService as any, - defaults.statusService as any, - defaults.notificationsService as any, defaults.checkService as any, defaults.settingsService as any, - defaults.buffer as any, - defaults.incidentService as any, - defaults.maintenanceWindowsRepository as any, defaults.monitorsRepository as any, defaults.teamsRepository as any, defaults.monitorStatsRepository as any, defaults.checksRepository as any, defaults.incidentsRepository as any, - defaults.geoChecksService as any, - defaults.geoChecksRepository as any + defaults.geoChecksRepository as any, + defaults.reactorDispatcher as any, + defaults.checkPipeline as any, + defaults.geoCheckPipeline as any ); - return { helper, defaults, maintenanceWindowsRepository }; + return { helper, defaults }; }; const makeMonitor = (overrides?: Partial): Monitor => @@ -94,7 +90,7 @@ const makeMonitor = (overrides?: Partial): Monitor => // ── Tests ──────────────────────────────────────────────────────────────────── -describe("QueueHelper", () => { +describe("WorkerHelper", () => { describe("serviceName", () => { it("returns JobQueueHelper", () => { const { helper } = createHelper(); @@ -102,278 +98,63 @@ describe("QueueHelper", () => { }); }); - // ── getHeartbeatJob ────────────────────────────────────────────────────── + // ── getHeartbeatJob (glue) ──────────────────────────────────────────────── describe("getHeartbeatJob", () => { - it("throws when monitor id is missing", async () => { - const { helper } = createHelper(); - const job = helper.getHeartbeatJob(); - await expect(job({} as Monitor)).rejects.toThrow("No monitor id"); + it("runs the pipeline and dispatches the evaluation to reactors", async () => { + const evaluation = { monitor: { id: "m1" }, decision: {} }; + const { helper, defaults } = createHelper({ checkPipeline: { run: jest.fn().mockResolvedValue(evaluation) } }); + + await helper.getHeartbeatJob()(makeMonitor()); + + expect(defaults.checkPipeline.run).toHaveBeenCalled(); + expect(defaults.reactorDispatcher.dispatch).toHaveBeenCalledWith(evaluation); }); - it("skips execution and sets maintenance status when in maintenance window", async () => { - const { helper, defaults } = createHelper(); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(true); - const job = helper.getHeartbeatJob(); + it("does not dispatch when the pipeline returns null (skipped)", async () => { + const { helper, defaults } = createHelper({ checkPipeline: { run: jest.fn().mockResolvedValue(null) } }); - await job(makeMonitor({ status: "up" })); + await helper.getHeartbeatJob()(makeMonitor()); - expect(defaults.networkService.requestStatus).not.toHaveBeenCalled(); - expect(defaults.monitorsRepository.updateById).toHaveBeenCalledWith("m1", "team", { status: "maintenance" }); + expect(defaults.reactorDispatcher.dispatch).not.toHaveBeenCalled(); }); - it("skips monitor status update when already in maintenance", async () => { - const { helper, defaults } = createHelper(); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(true); - const job = helper.getHeartbeatJob(); + it("logs and rethrows when the pipeline throws", async () => { + const { helper, defaults } = createHelper({ checkPipeline: { run: jest.fn().mockRejectedValue(new Error("boom")) } }); - await job(makeMonitor({ status: "maintenance" })); - - expect(defaults.monitorsRepository.updateById).not.toHaveBeenCalled(); + await expect(helper.getHeartbeatJob()(makeMonitor())).rejects.toThrow("boom"); + expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: "boom" })); }); - it("throws when network response is null", async () => { - const { helper } = createHelper({ - networkService: { requestStatus: jest.fn().mockResolvedValue(null) }, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatJob(); + it("rethrows non-Error failures", async () => { + const { helper } = createHelper({ checkPipeline: { run: jest.fn().mockRejectedValue("unexpected") } }); - await expect(job(makeMonitor())).rejects.toThrow("No network response"); - }); - - it("returns early and warns when buildCheck returns null", async () => { - const networkResponse = { monitorId: "m1", status: true, code: 200, message: "OK" }; - const { helper, defaults } = createHelper({ - networkService: { requestStatus: jest.fn().mockResolvedValue(networkResponse) }, - checkService: { buildCheck: jest.fn().mockReturnValue(null), deleteOlderThan: jest.fn() }, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatJob(); - - await job(makeMonitor()); - - expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("No check could be built") })); - expect(defaults.statusService.updateMonitorStatus).not.toHaveBeenCalled(); - }); - - it("processes full pipeline: network → check → status → incident", async () => { - const networkResponse = { monitorId: "m1", status: true, code: 200, message: "OK" }; - const { helper, defaults } = createHelper({ - networkService: { requestStatus: jest.fn().mockResolvedValue(networkResponse) }, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatJob(); - - await job(makeMonitor()); - - expect(defaults.buffer.addToBuffer).toHaveBeenCalled(); - expect(defaults.statusService.updateMonitorStatus).toHaveBeenCalled(); - expect(defaults.incidentService.handleIncident).toHaveBeenCalled(); - }); - - it("sends notifications when decision says shouldSendNotification", async () => { - const networkResponse = { monitorId: "m1", status: false, code: 500, message: "Error" }; - const statusResult = { monitor: { id: "m1", status: "down" }, statusChanged: true, prevStatus: "up", code: 500 }; - const { helper, defaults } = createHelper({ - networkService: { requestStatus: jest.fn().mockResolvedValue(networkResponse) }, - statusService: { updateMonitorStatus: jest.fn().mockResolvedValue(statusResult) }, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatJob(); - - await job(makeMonitor()); - - expect(defaults.notificationsService.handleNotifications).toHaveBeenCalledWith( - statusResult.monitor, - networkResponse, - expect.objectContaining({ shouldSendNotification: true, shouldCreateIncident: true }) - ); - }); - - it("logs error when notifications fail (fire-and-forget)", async () => { - const networkResponse = { monitorId: "m1", status: false, code: 500, message: "Error" }; - const statusResult = { monitor: { id: "m1", status: "down" }, statusChanged: true, prevStatus: "up", code: 500 }; - const notificationsService = { handleNotifications: jest.fn().mockRejectedValue(new Error("smtp down")) }; - const { helper, defaults } = createHelper({ - networkService: { requestStatus: jest.fn().mockResolvedValue(networkResponse) }, - statusService: { updateMonitorStatus: jest.fn().mockResolvedValue(statusResult) }, - notificationsService, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatJob(); - - await job(makeMonitor()); - - // Wait for the fire-and-forget catch to execute - await new Promise((r) => setTimeout(r, 10)); - expect(defaults.logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("smtp down") })); - }); - - it("logs error when notifications fail with non-Error (fire-and-forget)", async () => { - const networkResponse = { monitorId: "m1", status: false, code: 500, message: "Error" }; - const statusResult = { monitor: { id: "m1", status: "down" }, statusChanged: true, prevStatus: "up", code: 500 }; - const notificationsService = { handleNotifications: jest.fn().mockRejectedValue("string error") }; - const { helper, defaults } = createHelper({ - networkService: { requestStatus: jest.fn().mockResolvedValue(networkResponse) }, - statusService: { updateMonitorStatus: jest.fn().mockResolvedValue(statusResult) }, - notificationsService, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatJob(); - - await job(makeMonitor()); - await new Promise((r) => setTimeout(r, 10)); - expect(defaults.logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("Unknown error") })); - }); - - it("logs warning when incident handling fails (fire-and-forget)", async () => { - const networkResponse = { monitorId: "m1", status: true, code: 200, message: "OK" }; - const incidentService = { handleIncident: jest.fn().mockRejectedValue(new Error("db error")) }; - const { helper, defaults } = createHelper({ - networkService: { requestStatus: jest.fn().mockResolvedValue(networkResponse) }, - incidentService, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatJob(); - - await job(makeMonitor()); - await new Promise((r) => setTimeout(r, 10)); - expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("db error") })); - }); - - it("logs warning when incident handling fails with non-Error", async () => { - const networkResponse = { monitorId: "m1", status: true, code: 200, message: "OK" }; - const incidentService = { handleIncident: jest.fn().mockRejectedValue(42) }; - const { helper, defaults } = createHelper({ - networkService: { requestStatus: jest.fn().mockResolvedValue(networkResponse) }, - incidentService, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatJob(); - - await job(makeMonitor()); - await new Promise((r) => setTimeout(r, 10)); - expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("Unknown error") })); - }); - - it("logs and rethrows on unexpected error with non-Error", async () => { - const { helper } = createHelper({ - networkService: { requestStatus: jest.fn().mockRejectedValue("unexpected") }, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatJob(); - - await expect(job(makeMonitor())).rejects.toBe("unexpected"); + await expect(helper.getHeartbeatJob()(makeMonitor())).rejects.toBe("unexpected"); }); }); - // ── getHeartbeatGeoJob ─────────────────────────────────────────────────── + // ── getHeartbeatGeoJob (glue) ───────────────────────────────────────────── describe("getHeartbeatGeoJob", () => { - it("throws nothing when monitor id is missing (logs error)", async () => { + it("runs the geo pipeline", async () => { const { helper, defaults } = createHelper(); - const job = helper.getHeartbeatGeoJob(); - await job({} as Monitor); + await helper.getHeartbeatGeoJob()(makeMonitor()); - expect(defaults.logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "No monitor id" })); + expect(defaults.geoCheckPipeline.run).toHaveBeenCalled(); }); - it("returns early when geoCheckEnabled is false", async () => { - const { helper, defaults } = createHelper(); - const job = helper.getHeartbeatGeoJob(); - - await job(makeMonitor({ geoCheckEnabled: false })); - - expect(defaults.geoChecksService.buildGeoCheck).not.toHaveBeenCalled(); - }); - - it("returns early when monitor type does not support geo checks", async () => { - const { helper, defaults } = createHelper(); - const job = helper.getHeartbeatGeoJob(); - - await job(makeMonitor({ geoCheckEnabled: true, type: "hardware", geoCheckLocations: ["us-east"] })); - - expect(defaults.logger.debug).toHaveBeenCalledWith( - expect.objectContaining({ message: expect.stringContaining("does not support geo checks") }) - ); - expect(defaults.geoChecksService.buildGeoCheck).not.toHaveBeenCalled(); - }); - - it("returns early when geoCheckLocations is empty", async () => { - const { helper, defaults } = createHelper(); - const job = helper.getHeartbeatGeoJob(); - - await job(makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckLocations: [] })); - - expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("No geo check locations") })); - }); - - it("returns early when geoCheckLocations is undefined", async () => { - const { helper, defaults } = createHelper(); - const job = helper.getHeartbeatGeoJob(); - - await job(makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckLocations: undefined as any })); - - expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("No geo check locations") })); - }); - - it("skips when in maintenance window", async () => { - const { helper, defaults } = createHelper(); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(true); - const job = helper.getHeartbeatGeoJob(); - - await job(makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckLocations: ["us-east"] })); - - expect(defaults.logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("maintenance window") })); - expect(defaults.geoChecksService.buildGeoCheck).not.toHaveBeenCalled(); - }); - - it("warns when buildGeoCheck returns null", async () => { - const { helper, defaults } = createHelper(); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatGeoJob(); - - await job(makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckLocations: ["us-east"] })); - - expect(defaults.logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("No geo check could be built") })); - }); - - it("adds geo check to buffer on success", async () => { - const geoCheck = { id: "gc-1", monitorId: "m1" }; - const { helper, defaults } = createHelper({ - geoChecksService: { buildGeoCheck: jest.fn().mockResolvedValue(geoCheck) }, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatGeoJob(); - - await job(makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckLocations: ["us-east"] })); - - expect(defaults.buffer.addGeoCheckToBuffer).toHaveBeenCalledWith(geoCheck); - expect(defaults.logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("Geo check job executed") })); - }); - - it("logs error on unexpected failure without rethrowing", async () => { - const { helper, defaults } = createHelper({ - geoChecksService: { buildGeoCheck: jest.fn().mockRejectedValue(new Error("api timeout")) }, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatGeoJob(); - - await job(makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckLocations: ["us-east"] })); + it("catches and logs a pipeline error without rethrowing", async () => { + const { helper, defaults } = createHelper({ geoCheckPipeline: { run: jest.fn().mockRejectedValue(new Error("api timeout")) } }); + await expect(helper.getHeartbeatGeoJob()(makeMonitor())).resolves.toBeUndefined(); expect(defaults.logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "api timeout" })); }); - it("logs 'Unknown error' for non-Error exceptions", async () => { - const { helper, defaults } = createHelper({ - geoChecksService: { buildGeoCheck: jest.fn().mockRejectedValue("string error") }, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatGeoJob(); + it("logs 'Unknown error' for non-Error failures", async () => { + const { helper, defaults } = createHelper({ geoCheckPipeline: { run: jest.fn().mockRejectedValue("string error") } }); - await job(makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckLocations: ["us-east"] })); + await helper.getHeartbeatGeoJob()(makeMonitor()); expect(defaults.logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Unknown error", stack: undefined })); }); @@ -400,7 +181,6 @@ describe("QueueHelper", () => { expect(defaults.checksRepository.deleteByMonitorIdsNotIn).toHaveBeenCalledWith(["m1"]); expect(defaults.incidentsRepository.deleteByMonitorIdsNotIn).toHaveBeenCalledWith(["m1"]); expect(defaults.geoChecksRepository.deleteByMonitorIdsNotIn).toHaveBeenCalledWith(["m1"]); - // Logs info for each deleted count > 0 expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("2 orphaned monitors") })); expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("3 orphaned monitor stats") })); expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("4 orphaned checks") })); @@ -413,7 +193,6 @@ describe("QueueHelper", () => { const job = helper.getCleanupOrphanedJob(); await job(); - // Only start and completion logs, no deletion logs const infoMessages = (defaults.logger.info as jest.Mock).mock.calls.map((c: any) => c[0].message); expect(infoMessages).toContain("Starting cleanup of orphaned data"); expect(infoMessages).toContain("Cleanup of orphaned data completed"); @@ -483,236 +262,4 @@ describe("QueueHelper", () => { expect(defaults.logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Unknown error", stack: undefined })); }); }); - - // ── isInMaintenanceWindow ──────────────────────────────────────────────── - - describe("isInMaintenanceWindow", () => { - it("returns true when an active window spans now", async () => { - const now = new Date(); - const { helper, maintenanceWindowsRepository } = createHelper(); - maintenanceWindowsRepository.findByMonitorId.mockResolvedValue([ - { - active: true, - start: new Date(now.getTime() - 1000).toISOString(), - end: new Date(now.getTime() + 1000).toISOString(), - repeat: 0, - }, - ]); - await expect(helper.isInMaintenanceWindow("m1", "team")).resolves.toBe(true); - }); - - it("returns true when repeat interval advances window into current time", async () => { - const now = Date.now(); - const { helper, maintenanceWindowsRepository } = createHelper(); - maintenanceWindowsRepository.findByMonitorId.mockResolvedValue([ - { - active: true, - start: new Date(now - 7200000).toISOString(), - end: new Date(now - 6600000).toISOString(), - repeat: 3600000, - }, - ]); - await expect(helper.isInMaintenanceWindow("m1", "team")).resolves.toBe(true); - }); - - it("returns false when no active windows exist", async () => { - const { helper } = createHelper(); - await expect(helper.isInMaintenanceWindow("m1", "team")).resolves.toBe(false); - }); - - it("returns false when window is inactive", async () => { - const now = new Date(); - const { helper, maintenanceWindowsRepository } = createHelper(); - maintenanceWindowsRepository.findByMonitorId.mockResolvedValue([ - { - active: false, - start: new Date(now.getTime() - 1000).toISOString(), - end: new Date(now.getTime() + 1000).toISOString(), - repeat: 0, - }, - ]); - await expect(helper.isInMaintenanceWindow("m1", "team")).resolves.toBe(false); - }); - - it("returns false when active window is in the past without repeat", async () => { - const now = Date.now(); - const { helper, maintenanceWindowsRepository } = createHelper(); - maintenanceWindowsRepository.findByMonitorId.mockResolvedValue([ - { - active: true, - start: new Date(now - 5000).toISOString(), - end: new Date(now - 1000).toISOString(), - repeat: 0, - }, - ]); - await expect(helper.isInMaintenanceWindow("m1", "team")).resolves.toBe(false); - }); - - it("returns false when repeat advances past current time without match", async () => { - const now = Date.now(); - const { helper, maintenanceWindowsRepository } = createHelper(); - // Window was 2 hours ago, lasts 10 minutes, repeats every hour - // First repeat: -1h to -50min (past), second repeat: now to +10min... actually this would match - // Use a repeat that skips over now - maintenanceWindowsRepository.findByMonitorId.mockResolvedValue([ - { - active: true, - start: new Date(now - 3600000 * 3 - 300000).toISOString(), // 3h5min ago - end: new Date(now - 3600000 * 3).toISOString(), // 3h ago, 5min window - repeat: 3600000, // 1 hour repeat - }, - ]); - // Advances: -2h5m to -2h, -1h5m to -1h, -5m to 0 — that last one might match if now is within 5min - // Let's use a different setup that clearly doesn't match - maintenanceWindowsRepository.findByMonitorId.mockResolvedValue([ - { - active: true, - start: new Date(now - 1800000 - 60000).toISOString(), // 31 min ago - end: new Date(now - 1800000).toISOString(), // 30 min ago, 1 min window - repeat: 3600000, // 1 hour repeat — next window would be in 29 min - }, - ]); - await expect(helper.isInMaintenanceWindow("m1", "team")).resolves.toBe(false); - }); - - it("preserves previous true result via accumulator", async () => { - const now = new Date(); - const { helper, maintenanceWindowsRepository } = createHelper(); - maintenanceWindowsRepository.findByMonitorId.mockResolvedValue([ - { - active: true, - start: new Date(now.getTime() - 1000).toISOString(), - end: new Date(now.getTime() + 1000).toISOString(), - repeat: 0, - }, - { - active: false, - start: new Date(now.getTime() - 1000).toISOString(), - end: new Date(now.getTime() + 1000).toISOString(), - repeat: 0, - }, - ]); - // Second window is inactive but first already matched — accumulator should preserve true - await expect(helper.isInMaintenanceWindow("m1", "team")).resolves.toBe(true); - }); - }); - - // ── evaluateMonitorAction (tested indirectly via getHeartbeatJob) ───────── - - describe("evaluateMonitorAction", () => { - const runJobAndGetDecision = async (statusResult: Record) => { - const networkResponse = { monitorId: "m1", status: true, code: 200, message: "OK" }; - const incidentService = { handleIncident: jest.fn().mockResolvedValue(undefined) }; - const notificationsService = { handleNotifications: jest.fn().mockResolvedValue(undefined) }; - const { helper } = createHelper({ - networkService: { requestStatus: jest.fn().mockResolvedValue(networkResponse) }, - statusService: { updateMonitorStatus: jest.fn().mockResolvedValue(statusResult) }, - incidentService, - notificationsService, - }); - jest.spyOn(helper, "isInMaintenanceWindow").mockResolvedValue(false); - const job = helper.getHeartbeatJob(); - await job(makeMonitor()); - return { incidentService, notificationsService }; - }; - - it("does nothing when statusChanged is false", async () => { - const { incidentService, notificationsService } = await runJobAndGetDecision({ - monitor: { id: "m1", status: "up" }, - statusChanged: false, - prevStatus: "up", - code: 200, - }); - expect(incidentService.handleIncident).toHaveBeenCalledWith( - expect.anything(), - expect.anything(), - expect.objectContaining({ shouldCreateIncident: false, shouldResolveIncident: false, shouldSendNotification: false }), - expect.anything() - ); - }); - - it("creates incident and notifies when monitor goes down", async () => { - const { incidentService } = await runJobAndGetDecision({ - monitor: { id: "m1", status: "down" }, - statusChanged: true, - prevStatus: "up", - code: 500, - }); - expect(incidentService.handleIncident).toHaveBeenCalledWith( - expect.anything(), - expect.anything(), - expect.objectContaining({ - shouldCreateIncident: true, - shouldSendNotification: true, - incidentReason: "status_down", - notificationReason: "status_change", - }), - expect.anything() - ); - }); - - it("creates incident and notifies when monitor is breached", async () => { - const { incidentService } = await runJobAndGetDecision({ - monitor: { id: "m1", status: "breached" }, - statusChanged: true, - prevStatus: "up", - code: 200, - }); - expect(incidentService.handleIncident).toHaveBeenCalledWith( - expect.anything(), - expect.anything(), - expect.objectContaining({ - shouldCreateIncident: true, - incidentReason: "threshold_breach", - notificationReason: "threshold_breach", - }), - expect.anything() - ); - }); - - it("resolves incident when monitor recovers from down", async () => { - const { incidentService } = await runJobAndGetDecision({ - monitor: { id: "m1", status: "up" }, - statusChanged: true, - prevStatus: "down", - code: 200, - }); - expect(incidentService.handleIncident).toHaveBeenCalledWith( - expect.anything(), - expect.anything(), - expect.objectContaining({ shouldResolveIncident: true, shouldSendNotification: true }), - expect.anything() - ); - }); - - it("resolves incident when monitor recovers from breached", async () => { - const { incidentService } = await runJobAndGetDecision({ - monitor: { id: "m1", status: "up" }, - statusChanged: true, - prevStatus: "breached", - code: 200, - }); - expect(incidentService.handleIncident).toHaveBeenCalledWith( - expect.anything(), - expect.anything(), - expect.objectContaining({ shouldResolveIncident: true }), - expect.anything() - ); - }); - - it("does not create or resolve for unhandled status transitions", async () => { - const { incidentService } = await runJobAndGetDecision({ - monitor: { id: "m1", status: "paused" }, - statusChanged: true, - prevStatus: "up", - code: 200, - }); - expect(incidentService.handleIncident).toHaveBeenCalledWith( - expect.anything(), - expect.anything(), - expect.objectContaining({ shouldCreateIncident: false, shouldResolveIncident: false, shouldSendNotification: false }), - expect.anything() - ); - }); - }); });