diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 8d872201e..db6104b83 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -107,6 +107,8 @@ 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"; +import { CheckProducer } from "@/worker/worker.check-producer.js"; +import { CheckEvaluator } from "@/worker/worker.check-evaluator.js"; export type InitializedServices = { settingsService: ISettingsService; @@ -320,17 +322,12 @@ export const initializeServices = async ({ const incidentReactor = new IncidentReactor(incidentService); const reactorDispatcher = new ReactorDispatcher(logger, [notificationReactor, incidentReactor]); + // Check producer/evaluator + const checkProducer = new CheckProducer(monitorsRepository, maintenanceWindowsRepository, checkService, networkService, bufferService, logger); + const checkEvaluator = new CheckEvaluator(statusService, monitorStatusPolicy); + // pipelines - const checkPipeline = new CheckPipeline( - monitorsRepository, - maintenanceWindowsRepository, - checkService, - networkService, - bufferService, - monitorStatusPolicy, - statusService, - logger - ); + const checkPipeline = new CheckPipeline(checkProducer, checkEvaluator); const geoCheckPipeline = new GeoChecksPipeline(maintenanceWindowsRepository, geoChecksService, bufferService, logger); diff --git a/server/src/worker/worker.check-evaluator.ts b/server/src/worker/worker.check-evaluator.ts new file mode 100644 index 000000000..bbae96d69 --- /dev/null +++ b/server/src/worker/worker.check-evaluator.ts @@ -0,0 +1,33 @@ +import { MonitorStatusResponse } from "@/types/network.js"; +import { MonitorEvaluation } from "@/worker/worker.interface.js"; +import { Check } from "@/domain/checks/check.type.js"; +import { IMonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js"; +import { IStatusService } from "@/service/statusService.js"; + +export interface ICheckEvaluator { + evaluate(status: MonitorStatusResponse, check: Check): Promise; +} + +export class CheckEvaluator implements ICheckEvaluator { + constructor( + private statusService: IStatusService, + private monitorStatusPolicy: IMonitorStatusPolicy + ) {} + evaluate = async (status: MonitorStatusResponse, check: 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; + }; +} diff --git a/server/src/worker/worker.check-pipeline.ts b/server/src/worker/worker.check-pipeline.ts index c218b8fe8..c472c4bfa 100644 --- a/server/src/worker/worker.check-pipeline.ts +++ b/server/src/worker/worker.check-pipeline.ts @@ -1,17 +1,14 @@ 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"; +import { ICheckProducer } from "@/worker/worker.check-producer.js"; +import { ICheckEvaluator } from "@/worker/worker.check-evaluator.js"; const SERVICE_NAME = "CheckPipeline"; @@ -21,83 +18,16 @@ export interface ICheckPipeline { 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 checkProducer: ICheckProducer, + private checkEvaluator: ICheckEvaluator ) {} - 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" }); + const checkResult = await this.checkProducer.produce(monitor); + if (!checkResult) { + return null; // skipped due to maintenance / geo guards / other guards } - // **************************** - // 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.toCheck(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, - }; + const evaluation = await this.checkEvaluator.evaluate(checkResult.status, checkResult.check); return evaluation; }; } diff --git a/server/src/worker/worker.check-producer.ts b/server/src/worker/worker.check-producer.ts new file mode 100644 index 000000000..ad3975909 --- /dev/null +++ b/server/src/worker/worker.check-producer.ts @@ -0,0 +1,82 @@ +import { Monitor } from "@/domain/monitors/monitor.types.js"; +import { Check } from "@/domain/checks/check.type.js"; +import { MonitorStatusResponse } from "@/types/network.js"; +import { AppError } from "@/utils/AppError.js"; +import { ILogger } from "@/utils/logger.js"; +import { IBufferService } from "@/service/bufferService.js"; +import { INetworkService } from "@/service/networkService.js"; +import { ICheckService } from "@/domain/checks/check.service.js"; +import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; +import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js"; +import { isWindowActive } from "@/utils/maintenanceWindow.js"; + +export interface ICheckProducer { + produce(monitor: Monitor): Promise<{ status: MonitorStatusResponse; check: Check } | null>; +} + +const SERVICE_NAME = "CheckProducer"; + +export class CheckProducer implements ICheckProducer { + constructor( + private monitorsRepository: IMonitorsRepository, + private maintenanceWindowsRepository: IMaintenanceWindowsRepository, + private checkService: ICheckService, + private networkService: INetworkService, + private bufferService: IBufferService, + private logger: ILogger + ) {} + + private async isInMaintenanceWindow(monitorId: string, teamId: string) { + const windows = await this.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId); + const now = new Date(); + return windows.some((w) => isWindowActive(w, now)); + } + + produce = async (monitor: Monitor) => { + if (!monitor.id) { + throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "produce" }); + } + // **************************** + // 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: "produce", + }); + 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.toCheck(status); + if (!check) { + this.logger.warn({ + message: `No check could be built for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "produce", + details: { code: status.code, message: status.message }, + }); + return null; + } + // Step 2b: Add to buffer + this.bufferService.addToBuffer(check); + return { status, check }; + }; +} diff --git a/server/test/helpers/heartbeatTestHarness.ts b/server/test/helpers/heartbeatTestHarness.ts index 26dcbbe0f..88187ac15 100644 --- a/server/test/helpers/heartbeatTestHarness.ts +++ b/server/test/helpers/heartbeatTestHarness.ts @@ -2,6 +2,8 @@ import { jest } from "@jest/globals"; 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 { CheckProducer } from "../../src/worker/worker.check-producer.ts"; +import { CheckEvaluator } from "../../src/worker/worker.check-evaluator.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"; @@ -149,16 +151,16 @@ export function createHeartbeatTestHarness(): HeartbeatTestHarness { const maintenanceWindowsRepo = { findByMonitorId: jest.fn().mockResolvedValue([]) }; const geoChecksService = { buildGeoCheck: jest.fn() }; - const checkPipeline = new CheckPipeline( + const checkProducer = new CheckProducer( monitorsRepo as any, maintenanceWindowsRepo as any, checkService as any, networkService as any, bufferStub as any, - new MonitorStatusPolicy(), - statusService as any, logger ); + const checkEvaluator = new CheckEvaluator(statusService as any, new MonitorStatusPolicy()); + const checkPipeline = new CheckPipeline(checkProducer, checkEvaluator); const geoCheckPipeline = new GeoChecksPipeline(maintenanceWindowsRepo as any, geoChecksService as any, bufferStub as any, logger); const notificationReactor = new NotificationReactor(notificationsService as any); diff --git a/server/test/unit/services/checkPipeline.test.ts b/server/test/unit/services/checkPipeline.test.ts index 4a71d9592..9b1fffd8b 100644 --- a/server/test/unit/services/checkPipeline.test.ts +++ b/server/test/unit/services/checkPipeline.test.ts @@ -1,5 +1,7 @@ import { describe, expect, it, jest } from "@jest/globals"; import { CheckPipeline, GeoChecksPipeline } from "../../../src/worker/worker.check-pipeline.ts"; +import { CheckProducer } from "../../../src/worker/worker.check-producer.ts"; +import { CheckEvaluator } from "../../../src/worker/worker.check-evaluator.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"; @@ -37,16 +39,16 @@ const createCheckPipeline = (overrides?: Record) => { monitorStatusPolicy: new MonitorStatusPolicy(), ...overrides, }; - const pipeline = new CheckPipeline( + const checkProducer = new CheckProducer( 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 ); + const checkEvaluator = new CheckEvaluator(defaults.statusService as any, defaults.monitorStatusPolicy as any); + const pipeline = new CheckPipeline(checkProducer, checkEvaluator); return { pipeline, defaults }; };