diff --git a/server/src/service/infrastructure/JobQueues/IJobQueue.ts b/server/src/service/infrastructure/JobQueues/IJobQueue.ts new file mode 100644 index 000000000..a97c8eb6a --- /dev/null +++ b/server/src/service/infrastructure/JobQueues/IJobQueue.ts @@ -0,0 +1,52 @@ +import { Monitor } from "@/types/monitor.js"; + +export type QueueJobFailure = { + monitorId: string | number; + monitorUrl: string | null; + monitorType: string | null; + failedAt: number | null; + failCount: number; + failReason: string | null; +}; + +export type QueueMetrics = { + jobs: number; + activeJobs: number; + failingJobs: number; + jobsWithFailures: QueueJobFailure[]; + totalRuns: number; + totalFailures: number; +}; + +export type QueueJobSummary = { + monitorId: string | number; + monitorUrl: string | null; + monitorType: string | null; + monitorInterval: number | null; + monitorGeoInterval: number | null; + monitorActive: boolean | null; + active: boolean; + lockedAt: number | null; + runCount: number; + failCount: number; + failReason: string | null; + lastRunAt: number | null; + lastFinishedAt: number | null; + lastRunTook: number | null; + lastFailedAt: number | null; + repeat: number | null; +}; + +export interface IJobQueue { + readonly serviceName: string; + init(): Promise; + addJob(monitorId: string, monitor: Monitor): Promise; + deleteJob(monitor: Monitor): Promise; + pauseJob(monitor: Monitor): Promise; + resumeJob(monitor: Monitor): Promise; + updateJob(monitor: Monitor): Promise; + shutdown(): Promise; + getMetrics(): Promise; + getJobs(): Promise; + flushQueues(): Promise<{ success: boolean }>; +} diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts new file mode 100644 index 000000000..4ef57047f --- /dev/null +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -0,0 +1,355 @@ +import { IMonitorsRepository } from "@/repositories/index.js"; +import { ILogger } from "@/utils/logger.js"; +import { MongoStore, Scheduler, IJob } from "less-simple-scheduler"; +import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js"; +import { Monitor, supportsGeoCheck } from "@/types/monitor.js"; +import { type QueueMetrics, IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js"; +import { EnvConfig } from "@/service/system/settingsService.js"; +const SERVICE_NAME = "JobQueue"; + +// For discriminating job types +type MonitorJob = Omit & { data: Monitor }; +const MONITOR_TEMPLATES = new Set(["monitor-job", "geo-check-job"]); +const isMonitorJob = (job: IJob): job is MonitorJob => MONITOR_TEMPLATES.has(job.template); + +export class LessSimpleQueue implements IJobQueue { + static SERVICE_NAME = SERVICE_NAME; + + private logger: ILogger; + private helper: IQueueHelper; + private monitorsRepository: IMonitorsRepository; + private readonly scheduler: Scheduler; + + constructor(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) { + this.logger = logger; + this.helper = helper; + this.monitorsRepository = monitorsRepository; + this.scheduler = scheduler; + } + + get serviceName() { + return LessSimpleQueue.SERVICE_NAME; + } + + private registerListeners = () => { + this.scheduler.on("scheduler:start", () => { + this.logger.info({ + message: "Scheduler started", + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("scheduler:stop", () => { + this.logger.info({ + message: "Scheduler stopped", + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("scheduler:error", (workerId, error) => { + this.logger.error({ + message: `w${workerId} error: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + stack: error instanceof Error ? error.stack : undefined, + }); + }); + + this.scheduler.on("job:locked", (workerId, job) => { + this.logger.debug({ + message: `w${workerId} ${job.id} locked`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:abort", (workerId, job, reason) => { + this.logger.warn({ + message: `w${workerId} ${job.id} aborted: ${reason}`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:attempt", (workerId, job, attempt) => { + this.logger.debug({ + message: `w${workerId} ${job.id} attempt ${attempt}`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:complete", (workerId, job) => { + this.logger.debug({ + message: `w${workerId} ${job.id} completed successfully`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:exhausted", (workerId, job, error) => { + this.logger.error({ + message: `w${workerId} ${job.id} exhausted all retries: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + stack: error instanceof Error ? error.stack : undefined, + }); + }); + + this.scheduler.on("job:fail", (workerId, job, error, attempt) => { + this.logger.warn({ + message: `w${workerId} ${job.id} failed on attempt ${attempt}: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + stack: error instanceof Error ? error.stack : undefined, + }); + }); + + this.scheduler.on("job:start", (workerId, job) => { + this.logger.debug({ + message: `w${workerId} ${job.id} started`, + service: SERVICE_NAME, + }); + }); + }; + + static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, envSettings: EnvConfig) { + const store = new MongoStore({ + url: envSettings.dbConnectionString ?? "mongodb://localhost:27017/uptime_db", + }); + const scheduler = new Scheduler(store, { + concurrency: 50, + }); + const instance = new LessSimpleQueue(logger, helper, monitorsRepository, scheduler); + await instance.init(); + + return instance; + } + + init = async () => { + try { + this.registerListeners(); + await this.scheduler.start(); + this.scheduler.addTemplate("monitor-job", this.helper.getHeartbeatJob()); + this.scheduler.addTemplate("geo-check-job", this.helper.getHeartbeatGeoJob()); + this.scheduler.addTemplate("cleanup-orphaned", this.helper.getCleanupOrphanedJob()); + this.scheduler.addTemplate("cleanup-retention-job", this.helper.getCleanupRetentionJob()); + const monitors = await this.monitorsRepository.findAll(); + if (!monitors) { + return true; + } + for (const monitor of monitors) { + const randomOffset = Math.floor(Math.random() * 100); + setTimeout(() => { + this.addJob(monitor.id, monitor); + }, randomOffset); + } + + this.scheduler.addJob({ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true, upsert: true }); + this.scheduler.addJob({ id: "cleanup-retention", template: "cleanup-retention-job", active: true, repeat: 24 * 60 * 60 * 1000, upsert: true }); + + return true; + } catch (error: unknown) { + this.logger.error({ + message: `Failed to initialize SuperSimpleQueue: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + method: "init", + }); + return false; + } + }; + + addJob = async (monitorId: string, monitor: Monitor) => { + this.scheduler.addJob({ + id: monitorId, + template: "monitor-job", + repeat: monitor.interval, + active: monitor.isActive, + data: monitor, + upsert: true, + }); + + // Return early if we don't need geo checks + if (!supportsGeoCheck(monitor.type)) { + return; + } + + // Add geo check job if enabled for HTTP monitors + if (monitor.geoCheckEnabled) { + this.scheduler.addJob({ + id: `${monitorId}-geo`, + template: "geo-check-job", + repeat: monitor.geoCheckInterval, + active: monitor.isActive, + data: monitor, + upsert: true, + }); + } + }; + + deleteJob = async (monitor: Monitor) => { + this.scheduler.removeJob(monitor.id); + const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); + if (geoJob) await this.scheduler.removeJob(`${monitor.id}-geo`); + }; + + pauseJob = async (monitor: Monitor) => { + const result = await this.scheduler.pauseJob(monitor.id); + if (result === false) { + throw new Error("Failed to pause monitor"); + } + this.scheduler.updateJob(monitor.id, { data: monitor }); + + const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); + if (geoJob) { + const geoRes = await this.scheduler.pauseJob(`${monitor.id}-geo`); + if (geoRes === false) { + this.logger.error({ + message: `Failed to pause geo check job for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "pauseJob", + }); + } + this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor }); + } + + this.logger.debug({ + message: `Paused monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "pauseJob", + }); + }; + + resumeJob = async (monitor: Monitor) => { + const result = await this.scheduler.resumeJob(monitor.id); + if (result === false) { + throw new Error("Failed to resume monitor"); + } + this.scheduler.updateJob(monitor.id, { data: monitor }); + + const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); + if (geoJob) { + const geoRes = await this.scheduler.resumeJob(`${monitor.id}-geo`); + if (geoRes === false) { + this.logger.error({ + message: `Failed to resume geo check job for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "resumeJob", + }); + } + this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor }); + } + this.logger.debug({ + message: `Resumed monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "resumeJob", + }); + }; + + private syncGeoJob = async (monitor: Monitor) => { + const geoJobId = `${monitor.id}-geo`; + const existingGeoJob = await this.scheduler.getJob(geoJobId); + + // If geoChecks have been disabled, or the monitor type doesn't support them, remove + if (!monitor.geoCheckEnabled || !supportsGeoCheck(monitor.type)) { + if (existingGeoJob) this.scheduler.removeJob(geoJobId); + return; + } + + // If the job exists, update it + if (existingGeoJob) { + this.scheduler.updateJob(geoJobId, { repeat: monitor.geoCheckInterval, active: monitor.isActive, data: monitor }); + return; + } + + // Otherwise, create it + this.scheduler.addJob({ + id: geoJobId, + template: "geo-check-job", + repeat: monitor.geoCheckInterval, + active: monitor.isActive, + data: monitor, + upsert: true, + }); + }; + + updateJob = async (monitor: Monitor) => { + this.scheduler.updateJob(monitor.id, { repeat: monitor.interval, data: monitor }); + await this.syncGeoJob(monitor); + }; + + shutdown = async () => { + this.scheduler.stop(); + }; + + getMetrics = async () => { + const jobs = await this.scheduler.getJobs(); + const metrics = jobs.reduce( + (acc, job) => { + const runCount = job.runCount ?? 0; + const failCount = job.failCount ?? 0; + const lastFailedAt = job.lastFailedAt ?? 0; + const lastRunAt = job.lastRunAt ?? 0; + acc.totalRuns += runCount; + acc.totalFailures += failCount; + acc.jobs++; + if (failCount > 0 && lastFailedAt >= lastRunAt) { + acc.failingJobs++; + } + + if (job.lockedAt) { + acc.activeJobs++; + } + + if (failCount > 0) { + const monitor = isMonitorJob(job) ? job.data : undefined; + acc.jobsWithFailures.push({ + monitorId: job.id, + monitorUrl: monitor?.url || null, + monitorType: monitor?.type || null, + failedAt: job.lastFailedAt ?? null, + failCount, + failReason: job.lastError ?? null, + }); + } + return acc; + }, + { + jobs: 0, + activeJobs: 0, + failingJobs: 0, + jobsWithFailures: [], + totalRuns: 0, + totalFailures: 0, + } + ); + return metrics; + }; + + getJobs = async () => { + const jobs = await this.scheduler.getJobs(); + return jobs.map((job) => { + const monitor = isMonitorJob(job) ? job.data : undefined; + return { + monitorId: job.id, + monitorUrl: monitor?.url || null, + monitorType: monitor?.type || null, + monitorInterval: monitor?.interval || null, + monitorGeoInterval: monitor?.geoCheckInterval || null, + monitorActive: monitor?.isActive ?? null, + active: job.active, + repeat: job.repeat ?? null, + lockedAt: job.lockedAt ?? null, + runCount: job.runCount ?? 0, + failCount: job.failCount ?? 0, + failReason: job.lastError ?? null, + lastRunAt: job.lastRunAt ?? null, + lastFinishedAt: job.lastRunAt ?? null, + lastRunTook: job.lockedAt ? null : (job.lastRunAt ?? 0) - (job.lastRunAt ?? 0), + lastFailedAt: job.lastFailedAt ?? null, + }; + }); + }; + + flushQueues = async () => { + const stopRes = await this.scheduler.stop(); + const flushRes = await this.scheduler.flushJobs(); + const initRes = await this.init(); + return { + success: Boolean(stopRes && flushRes && initRes), + }; + }; +} diff --git a/server/src/service/infrastructure/JobQueues/QueueHelper.ts b/server/src/service/infrastructure/JobQueues/QueueHelper.ts new file mode 100644 index 000000000..e162e8d94 --- /dev/null +++ b/server/src/service/infrastructure/JobQueues/QueueHelper.ts @@ -0,0 +1,437 @@ +const SERVICE_NAME = "JobQueueHelper"; +import type { Monitor } from "@/types/monitor.js"; +import { supportsGeoCheck } from "@/types/monitor.js"; +import { AppError } from "@/utils/AppError.js"; +import { + ICheckService, + INetworkService, + INotificationsService, + ISettingsService, + IStatusService, + IncidentService, + type IGeoChecksService, +} from "@/service/index.js"; +import { CHECK_TTL_SENTINEL, type StatusChangeResult } from "@/types/index.js"; +import { isWindowActive } from "@/utils/maintenanceWindow.js"; +import { + IMaintenanceWindowsRepository, + IMonitorsRepository, + ITeamsRepository, + IMonitorStatsRepository, + IChecksRepository, + IIncidentsRepository, + IGeoChecksRepository, +} from "@/repositories/index.js"; +import { ILogger } from "@/utils/logger.js"; +import { IBufferService } from "@/service/index.js"; + +export interface IQueueHelper { + readonly serviceName: string; + getHeartbeatJob(): (monitor: Monitor) => Promise; + getHeartbeatGeoJob(): (monitor: Monitor) => Promise; + getCleanupOrphanedJob(): () => Promise; + getCleanupRetentionJob(): () => Promise; + isInMaintenanceWindow(monitorId: string, teamId: string): Promise; +} + +export interface MonitorActionDecision { + shouldCreateIncident: boolean; + shouldResolveIncident: boolean; + shouldSendNotification: boolean; + incidentReason: "status_down" | "threshold_breach" | null; + notificationReason: "status_change" | "threshold_breach" | null; + thresholdBreaches?: { + cpu?: boolean; + memory?: boolean; + disk?: boolean; + temp?: boolean; + }; +} + +export class QueueHelper implements IQueueHelper { + 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; + + 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 + ) { + 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; + } + + get serviceName() { + return SuperSimpleQueueHelper.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, + }); + } + } catch (error: unknown) { + this.logger.warn({ + message: error instanceof Error ? error.message : "Unknown error", + service: SERVICE_NAME, + method: "getMonitorJob", + stack: error instanceof Error ? error.stack : undefined, + }); + throw error; + } + }; + }; + + getCleanupOrphanedJob = () => { + return async () => { + try { + this.logger.info({ + message: "Starting cleanup of orphaned data", + service: SERVICE_NAME, + method: "getCleanupOrphanedJob", + }); + + // Get all valid team IDs + const validTeamIds = await this.teamsRepository.findAllTeamIds(); + this.logger.debug({ + message: `Found ${validTeamIds.length} valid teams`, + service: SERVICE_NAME, + method: "getCleanupOrphanedJob", + }); + + // Remove orphaned monitors (monitors without a valid team) + const deletedMonitorCount = await this.monitorsRepository.deleteByTeamIdsNotIn(validTeamIds); + if (deletedMonitorCount > 0) { + this.logger.info({ + message: `Deleted ${deletedMonitorCount} orphaned monitors`, + service: SERVICE_NAME, + method: "getCleanupOrphanedJob", + }); + } + + // Remove orphaned monitorStats (stats without a valid monitor) + const allMonitorIds = await this.monitorsRepository.findAllMonitorIds(); + this.logger.debug({ + message: `Found ${allMonitorIds.length} valid monitors`, + service: SERVICE_NAME, + method: "getCleanupOrphanedJob", + }); + + const deletedStatsCount = await this.monitorStatsRepository.deleteByMonitorIdsNotIn(allMonitorIds); + if (deletedStatsCount > 0) { + this.logger.info({ + message: `Deleted ${deletedStatsCount} orphaned monitor stats`, + service: SERVICE_NAME, + method: "getCleanupOrphanedJob", + }); + } + + // Remove orphaned checks + const deletedChecksCount = await this.checksRepository.deleteByMonitorIdsNotIn(allMonitorIds); + if (deletedChecksCount > 0) { + this.logger.info({ + message: `Deleted ${deletedChecksCount} orphaned checks`, + service: SERVICE_NAME, + method: "getCleanupOrphanedJob", + }); + } + + // Remove orphaned incidents + const deletedIncidentsCount = await this.incidentsRepository.deleteByMonitorIdsNotIn(allMonitorIds); + if (deletedIncidentsCount > 0) { + this.logger.info({ + message: `Deleted ${deletedIncidentsCount} orphaned incidents`, + service: SERVICE_NAME, + method: "getCleanupOrphanedJob", + }); + } + + // Remove orphaned geo checks + const deletedGeoChecksCount = await this.geoChecksRepository.deleteByMonitorIdsNotIn(allMonitorIds); + if (deletedGeoChecksCount > 0) { + this.logger.info({ + message: `Deleted ${deletedGeoChecksCount} orphaned geo checks`, + service: SERVICE_NAME, + method: "getCleanupOrphanedJob", + }); + } + + this.logger.info({ + message: "Cleanup of orphaned data completed", + service: SERVICE_NAME, + method: "getCleanupOrphanedJob", + }); + } catch (error: unknown) { + this.logger.warn({ + message: error instanceof Error ? error.message : "Unknown error", + service: SERVICE_NAME, + method: "getCleanupOrphanedJob", + stack: error instanceof Error ? error.stack : undefined, + }); + throw error; + } + }; + }; + + 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 { + const settings = await this.settingsService.getDBSettings(); + + const checkTTL = settings.checkTTL; // Check TTL is in DAYS, not MS + + if (checkTTL === CHECK_TTL_SENTINEL) { + this.logger.info({ + message: `Check TTL is set to unlimited, skipping cleanup`, + service: SERVICE_NAME, + method: "getCleanupRetentionJob", + }); + return; + } + const checkTTLInMs = checkTTL * 24 * 60 * 60 * 1000; + const cutoffDate = new Date(Date.now() - checkTTLInMs); + const deleteCount = await this.checkService.deleteOlderThan(cutoffDate); + this.logger.info({ + message: `Deleted ${deleteCount} checks older than ${cutoffDate.toISOString()}`, + service: SERVICE_NAME, + method: "getCleanupRetentionJob", + }); + } catch (error: unknown) { + this.logger.error({ + message: error instanceof Error ? error.message : "Unknown error", + service: SERVICE_NAME, + method: "getCleanupRetentionJob", + stack: error instanceof Error ? error.stack : undefined, + }); + } + }; + }; + + 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/service/infrastructure/JobQueues/SuperSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts new file mode 100644 index 000000000..b98ecb2db --- /dev/null +++ b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts @@ -0,0 +1,395 @@ +import { IMonitorsRepository } from "@/repositories/index.js"; +import { ILogger } from "@/utils/logger.js"; +import Scheduler from "super-simple-scheduler"; +import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js"; +import { Monitor, MonitorType, supportsGeoCheck } from "@/types/monitor.js"; +const SERVICE_NAME = "JobQueue"; + +type QueueJobFailure = { + monitorId: string | number; + monitorUrl: string | null; + monitorType: string | null; + failedAt: number | null; + failCount: number; + failReason: string | null; +}; + +type QueueMetrics = { + jobs: number; + activeJobs: number; + failingJobs: number; + jobsWithFailures: QueueJobFailure[]; + totalRuns: number; + totalFailures: number; +}; + +type QueueJobSummary = { + monitorId: string | number; + monitorUrl: string | null; + monitorType: string | null; + monitorInterval: number | null; + monitorGeoInterval: number | null; + monitorActive: boolean | null; + active: boolean; + lockedAt: number | null; + runCount: number; + failCount: number; + failReason: string | null; + lastRunAt: number | null; + lastFinishedAt: number | null; + lastRunTook: number | null; + lastFailedAt: number | null; + repeat: number | null; +}; + +export interface ISuperSimpleQueue { + readonly serviceName: string; + init(): Promise; + addJob(monitorId: string, monitor: Monitor): Promise; + deleteJob(monitor: Monitor): Promise; + pauseJob(monitor: Monitor): Promise; + resumeJob(monitor: Monitor): Promise; + updateJob(monitor: Monitor): Promise; + shutdown(): Promise; + getMetrics(): Promise; + getJobs(): Promise; + flushQueues(): Promise<{ success: boolean }>; +} + +export class SuperSimpleQueue implements ISuperSimpleQueue { + static SERVICE_NAME = SERVICE_NAME; + + private logger: ILogger; + private helper: IQueueHelper; + private monitorsRepository: IMonitorsRepository; + private readonly scheduler: Scheduler; + + constructor(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) { + this.logger = logger; + this.helper = helper; + this.monitorsRepository = monitorsRepository; + this.scheduler = scheduler; + } + + get serviceName() { + return SuperSimpleQueue.SERVICE_NAME; + } + + private registerListeners = () => { + this.scheduler.on("scheduler:start", () => { + this.logger.info({ + message: "Scheduler started", + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("scheduler:stop", () => { + this.logger.info({ + message: "Scheduler stopped", + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("scheduler:error", (error) => { + this.logger.error({ + message: `Scheduler error: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + stack: error instanceof Error ? error.stack : undefined, + }); + }); + + this.scheduler.on("job:abort", (job, reason) => { + this.logger.warn({ + message: `${job.id} aborted: ${reason}`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:attempt", (job, attempt) => { + this.logger.debug({ + message: `${job.id} attempt ${attempt}`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:complete", (job) => { + this.logger.debug({ + message: `${job.id} completed successfully`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:exhausted", (job, error) => { + this.logger.error({ + message: `${job.id} exhausted all retries: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + stack: error instanceof Error ? error.stack : undefined, + }); + }); + + this.scheduler.on("job:fail", (job, error, attempt) => { + this.logger.warn({ + message: `${job.id} failed on attempt ${attempt}: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + stack: error instanceof Error ? error.stack : undefined, + }); + }); + + this.scheduler.on("job:start", (job) => { + this.logger.debug({ + message: `${job.id} started`, + service: SERVICE_NAME, + }); + }); + }; + + static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository) { + const scheduler = new Scheduler({ + concurrency: 50, + // storeType: "mongo", + // storeType: "redis", + // dbUri: envSettings.dbConnectionString, + }); + const instance = new SuperSimpleQueue(logger, helper, monitorsRepository, scheduler); + await instance.init(); + + return instance; + } + + init = async () => { + try { + this.registerListeners(); + + this.scheduler.start(); + this.scheduler.addTemplate("monitor-job", this.helper.getHeartbeatJob()); + this.scheduler.addTemplate("geo-check-job", this.helper.getHeartbeatGeoJob()); + this.scheduler.addTemplate("cleanup-orphaned", this.helper.getCleanupOrphanedJob()); + this.scheduler.addTemplate("cleanup-retention-job", this.helper.getCleanupRetentionJob()); + const monitors = await this.monitorsRepository.findAll(); + if (!monitors) { + return true; + } + for (const monitor of monitors) { + const randomOffset = Math.floor(Math.random() * 100); + setTimeout(() => { + this.addJob(monitor.id, monitor); + }, randomOffset); + } + + this.scheduler.addJob({ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true }); + this.scheduler.addJob({ id: "cleanup-retention", template: "cleanup-retention-job", active: true, repeat: 24 * 60 * 60 * 1000 }); + + return true; + } catch (error: unknown) { + this.logger.error({ + message: `Failed to initialize SuperSimpleQueue: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + method: "init", + }); + return false; + } + }; + + addJob = async (monitorId: string, monitor: Monitor) => { + this.scheduler.addJob({ + id: monitorId, + template: "monitor-job", + repeat: monitor.interval, + active: monitor.isActive, + data: monitor, + }); + + // Return early if we don't need geo checks + if (!supportsGeoCheck(monitor.type)) { + return; + } + + // Add geo check job if enabled for HTTP monitors + if (monitor.geoCheckEnabled) { + this.scheduler.addJob({ + id: `${monitorId}-geo`, + template: "geo-check-job", + repeat: monitor.geoCheckInterval, + active: monitor.isActive, + data: monitor, + }); + } + }; + + deleteJob = async (monitor: Monitor) => { + this.scheduler.removeJob(monitor.id); + const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); + if (geoJob) await this.scheduler.removeJob(`${monitor.id}-geo`); + }; + + pauseJob = async (monitor: Monitor) => { + const result = await this.scheduler.pauseJob(monitor.id); + if (result === false) { + throw new Error("Failed to pause monitor"); + } + this.scheduler.updateJob(monitor.id, { data: monitor }); + + const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); + if (geoJob) { + const geoRes = await this.scheduler.pauseJob(`${monitor.id}-geo`); + if (geoRes === false) { + this.logger.error({ + message: `Failed to pause geo check job for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "pauseJob", + }); + } + this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor }); + } + + this.logger.debug({ + message: `Paused monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "pauseJob", + }); + }; + + resumeJob = async (monitor: Monitor) => { + const result = await this.scheduler.resumeJob(monitor.id); + if (result === false) { + throw new Error("Failed to resume monitor"); + } + this.scheduler.updateJob(monitor.id, { data: monitor }); + + const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); + if (geoJob) { + const geoRes = await this.scheduler.resumeJob(`${monitor.id}-geo`); + if (geoRes === false) { + this.logger.error({ + message: `Failed to resume geo check job for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "resumeJob", + }); + } + this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor }); + } + this.logger.debug({ + message: `Resumed monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "resumeJob", + }); + }; + + private syncGeoJob = async (monitor: Monitor) => { + const geoJobId = `${monitor.id}-geo`; + const existingGeoJob = await this.scheduler.getJob(geoJobId); + + // If geoChecks have been disabled, or the monitor type doesn't support them, remove + if (!monitor.geoCheckEnabled || !supportsGeoCheck(monitor.type)) { + if (existingGeoJob) this.scheduler.removeJob(geoJobId); + return; + } + + // If the job exists, update it + if (existingGeoJob) { + this.scheduler.updateJob(geoJobId, { repeat: monitor.geoCheckInterval, active: monitor.isActive, data: monitor }); + return; + } + + // Otherwise, create it + this.scheduler.addJob({ + id: geoJobId, + template: "geo-check-job", + repeat: monitor.geoCheckInterval, + active: monitor.isActive, + data: monitor, + }); + }; + + updateJob = async (monitor: Monitor) => { + this.scheduler.updateJob(monitor.id, { repeat: monitor.interval, data: monitor }); + await this.syncGeoJob(monitor); + }; + + shutdown = async () => { + this.scheduler.stop(); + }; + + getMetrics = async () => { + const jobs = await this.scheduler.getJobs(); + const metrics = jobs.reduce( + (acc, job) => { + const runCount = job.runCount ?? 0; + const failCount = job.failCount ?? 0; + const lastFailedAt = job.lastFailedAt ?? 0; + const lastRunAt = job.lastRunAt ?? 0; + acc.totalRuns += runCount; + acc.totalFailures += failCount; + acc.jobs++; + if (failCount > 0 && lastFailedAt >= lastRunAt) { + acc.failingJobs++; + } + + if (job.lockedAt) { + acc.activeJobs++; + } + + if (failCount > 0) { + acc.jobsWithFailures.push({ + monitorId: job.id, + monitorUrl: job.data?.url || null, + monitorType: job.data?.type || null, + failedAt: job.lastFailedAt ?? null, + failCount, + failReason: job.lastFailReason ?? null, + }); + } + return acc; + }, + { + jobs: 0, + activeJobs: 0, + failingJobs: 0, + jobsWithFailures: [] as Array<{ + monitorId: string | number; + monitorUrl: string; + monitorType: MonitorType; + failedAt: number | null; + failCount: number; + failReason: string | null; + }>, + totalRuns: 0, + totalFailures: 0, + } + ); + return metrics; + }; + + getJobs = async () => { + const jobs = await this.scheduler.getJobs(); + return jobs.map((job) => { + return { + monitorId: job.id, + monitorUrl: job.data?.url || null, + monitorType: job.data?.type || null, + monitorInterval: job.data?.interval || null, + monitorGeoInterval: job.data?.geoCheckInterval || null, + monitorActive: job.data?.isActive ?? null, + active: job.active, + repeat: job.repeat ?? null, + lockedAt: job.lockedAt ?? null, + runCount: job.runCount ?? 0, + failCount: job.failCount ?? 0, + failReason: job.lastFailReason ?? null, + lastRunAt: job.lastRunAt ?? null, + lastFinishedAt: job.lastFinishedAt ?? null, + lastRunTook: job.lockedAt ? null : (job.lastFinishedAt ?? 0) - (job.lastRunAt ?? 0), + lastFailedAt: job.lastFailedAt ?? null, + }; + }); + }; + + flushQueues = async () => { + const stopRes = await this.scheduler.stop(); + const flushRes = await this.scheduler.flushJobs(); + const initRes = await this.init(); + return { + success: Boolean(stopRes && flushRes && initRes), + }; + }; +}