const cron = require('node-cron'); const db = require('./database'); const monitorService = require('./monitorService'); const notificationService = require('./notificationService'); const AGENT_CPU_THRESHOLD = 90; const AGENT_MEM_THRESHOLD = 90; const AGENT_DISK_THRESHOLD = 90; const AGENT_OFFLINE_MS = 3 * 60 * 1000; // matches UI isOnline() check class Scheduler { constructor() { this.jobs = new Map(); this.isRunning = false; this.lastCheckedAt = new Map(); this.slowAlertCooldown = new Map(); this.slowConsecutiveCount = new Map(); // Agent health tracking — keyed by agent id this.agentOnlineState = new Map(); // id -> boolean this.agentAlertState = new Map(); // id -> { cpu, mem, disk } } start() { // Run checks every minute cron.schedule('* * * * *', async () => { if (!this.isRunning) { this.isRunning = true; try { await this.runChecks(); } finally { this.isRunning = false; } } }); // Agent health check every 2 minutes cron.schedule('*/2 * * * *', async () => { await this.checkAgents(); }); // Archive old checks + update baselines + check escalations daily at 2 AM cron.schedule('0 2 * * *', () => { this.archiveOldChecks(); this.updateBaselines(); }); // Escalation check runs every hour cron.schedule('0 * * * *', async () => { await this.checkEscalations(); }); // Also run archive on startup (with delay to let DB init) setTimeout(() => { this.archiveOldChecks(); this.updateBaselines(); }, 5000); } async checkAgents() { try { const rows = db.prepare(` SELECT a.id, a.name, a.hostname, a.ip_address, a.os_info, a.last_seen_at, m.cpu_percent, m.mem_percent, m.disk_data FROM agents a LEFT JOIN agent_metrics m ON m.id = ( SELECT id FROM agent_metrics WHERE agent_id = a.id ORDER BY recorded_at DESC LIMIT 1 ) ORDER BY a.id `).all(); const liveIds = new Set(); for (const agent of rows) { liveIds.add(agent.id); const isOnline = agent.last_seen_at ? Date.now() - new Date(agent.last_seen_at).getTime() < AGENT_OFFLINE_MS : false; const wasOnline = this.agentOnlineState.get(agent.id); if (wasOnline === undefined) { // First poll — seed state without alerting this.agentOnlineState.set(agent.id, isOnline); this.agentAlertState.set(agent.id, { cpu: false, mem: false, disk: false }); continue; } // Online/offline transitions if (wasOnline && !isOnline) { await notificationService.sendAgentAlert( agent, 'offline', `Agent "${agent.name}" (${agent.hostname || agent.ip_address || 'unknown'}) has gone offline.` ); } else if (!wasOnline && isOnline) { await notificationService.sendAgentAlert( agent, 'online', `Agent "${agent.name}" (${agent.hostname || agent.ip_address || 'unknown'}) is back online.` ); this.agentAlertState.set(agent.id, { cpu: false, mem: false, disk: false }); } this.agentOnlineState.set(agent.id, isOnline); if (!isOnline || agent.cpu_percent == null) continue; const as = this.agentAlertState.get(agent.id) || { cpu: false, mem: false, disk: false }; // CPU if (agent.cpu_percent > AGENT_CPU_THRESHOLD && !as.cpu) { await notificationService.sendAgentAlert( agent, 'cpu', `Agent "${agent.name}" CPU is at ${Number(agent.cpu_percent).toFixed(1)}% (threshold: ${AGENT_CPU_THRESHOLD}%).` ); as.cpu = true; } else if (agent.cpu_percent <= AGENT_CPU_THRESHOLD) { as.cpu = false; } // Memory if (agent.mem_percent > AGENT_MEM_THRESHOLD && !as.mem) { await notificationService.sendAgentAlert( agent, 'memory', `Agent "${agent.name}" memory is at ${Number(agent.mem_percent).toFixed(1)}% (threshold: ${AGENT_MEM_THRESHOLD}%).` ); as.mem = true; } else if (agent.mem_percent <= AGENT_MEM_THRESHOLD) { as.mem = false; } // Disk — check all partitions if (agent.disk_data) { try { const disks = JSON.parse(agent.disk_data); const worst = disks.reduce((m, d) => (d.percent > (m?.percent ?? 0) ? d : m), null); if (worst && worst.percent > AGENT_DISK_THRESHOLD && !as.disk) { await notificationService.sendAgentAlert( agent, 'disk', `Agent "${agent.name}" disk ${worst.path} is at ${worst.percent}% (threshold: ${AGENT_DISK_THRESHOLD}%).` ); as.disk = true; } else if (!worst || worst.percent <= AGENT_DISK_THRESHOLD) { as.disk = false; } } catch (_) {} } this.agentAlertState.set(agent.id, as); } // Remove state for deleted agents for (const id of this.agentOnlineState.keys()) { if (!liveIds.has(id)) { this.agentOnlineState.delete(id); this.agentAlertState.delete(id); } } } catch (err) { console.error('Agent health check error:', err.message); } } archiveOldChecks() { try { const retentionSetting = db.prepare(`SELECT value FROM settings WHERE key = 'retention_days'`).get(); const globalRetentionDays = retentionSetting ? parseInt(retentionSetting.value) : 30; const resources = db.prepare(`SELECT id, retention_days FROM resources`).all(); const bulkArchive = db.prepare(` INSERT INTO archived_checks (resource_id, status, response_time, status_code, error_message, details, checked_at) SELECT resource_id, status, response_time, status_code, error_message, details, checked_at FROM checks WHERE resource_id = ? AND checked_at < ? `); const bulkDelete = db.prepare(`DELETE FROM checks WHERE resource_id = ? AND checked_at < ?`); const transaction = db.transaction(() => { resources.forEach(resource => { const retentionDays = resource.retention_days || globalRetentionDays; const cutoffDate = new Date(); cutoffDate.setDate(cutoffDate.getDate() - retentionDays); const cutoffString = cutoffDate.toISOString().replace('T', ' ').split('.')[0]; bulkArchive.run(resource.id, cutoffString); bulkDelete.run(resource.id, cutoffString); }); }); transaction(); } catch (error) { // Archive error handled silently } } updateBaselines() { try { const resources = db.prepare(`SELECT id, type FROM resources WHERE type != 'heartbeat'`).all(); for (const resource of resources) { const baseline = monitorService.calculateResponseTimeBaseline(resource.id, 7); if (baseline !== null) { db.prepare(`UPDATE resources SET response_time_baseline = ? WHERE id = ?`).run(baseline, resource.id); } } } catch (error) { // Baseline update error handled silently } } async checkEscalations() { try { const escalationSetting = db.prepare(`SELECT value FROM settings WHERE key = 'escalation_hours'`).get(); const escalationHours = parseInt(escalationSetting?.value || '4'); const fallbackSetting = db.prepare(`SELECT value FROM settings WHERE key = 'fallback_webhook'`).get(); const fallbackWebhook = fallbackSetting?.value || process.env.FALLBACK_WEBHOOK || ''; if (!fallbackWebhook) return; const cutoff = new Date(Date.now() - escalationHours * 60 * 60 * 1000) .toISOString().replace('T', ' ').split('.')[0]; const stalled = db.prepare(` SELECT i.*, r.name AS resource_name, r.url AS resource_url, r.type AS resource_type, r.maintenance_mode FROM incidents i JOIN resources r ON r.id = i.resource_id WHERE i.resolved_at IS NULL AND i.escalated = 0 AND i.started_at < ? AND r.maintenance_mode = 0 `).all(cutoff); for (const incident of stalled) { await notificationService.sendEscalationAlert(incident, fallbackWebhook); db.prepare(`UPDATE incidents SET escalated = 1 WHERE id = ?`).run(incident.id); } } catch (error) { // Escalation check error handled silently } } async runChecks() { const resources = db.prepare(`SELECT * FROM resources WHERE enabled = 1`).all(); // Read global defaults once per cycle so per-resource nulls inherit correctly const globalIntervalRow = db.prepare(`SELECT value FROM settings WHERE key = 'check_interval'`).get(); const globalTimeoutRow = db.prepare(`SELECT value FROM settings WHERE key = 'timeout'`).get(); const globalCheckInterval = parseInt(globalIntervalRow?.value) || 60000; const globalTimeout = parseInt(globalTimeoutRow?.value) || 5000; const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); const jitter = (maxMs = 1500) => Math.floor(Math.random() * maxMs); const baseSpacingMs = 250; // Collect incident alerts to flush as a group after all checks complete const pendingAlerts = []; const tasks = resources.map((resource, index) => (async () => { // Honour per-resource check_interval, falling back to global setting const now = Date.now(); const lastChecked = this.lastCheckedAt.get(resource.id) || 0; const interval = resource.check_interval || globalCheckInterval; if (now - lastChecked < interval) return; this.lastCheckedAt.set(resource.id, now); // Resolve effective timeout: per-resource override or global default const effectiveResource = resource.timeout ? resource : { ...resource, timeout: globalTimeout }; await sleep(index * baseSpacingMs + jitter()); try { const result = await monitorService.checkResource(effectiveResource); monitorService.saveCheck(result); // Rolling cap per resource (configurable via settings; default 10000) try { const capSetting = db.prepare(`SELECT value FROM settings WHERE key = 'checks_cap_per_resource'`).get(); const cap = capSetting ? parseInt(capSetting.value) : 10000; if (Number.isFinite(cap) && cap > 0) { const rowCount = db.prepare('SELECT COUNT(*) as cnt FROM checks WHERE resource_id = ?').get(resource.id).cnt; const excess = rowCount - cap; if (excess > 0) { db.prepare(` DELETE FROM checks WHERE id IN ( SELECT id FROM checks WHERE resource_id = ? ORDER BY id ASC LIMIT ? ) `).run(resource.id, excess); } } } catch (e) { // Non-fatal } const incident = monitorService.handleIncident(resource.id, result.status === 'down'); if (incident.type !== 'none' && !resource.maintenance_mode) { const stats = monitorService.getResourceStats(resource.id, 24); const recentChecks = db.prepare(` SELECT response_time, status FROM checks WHERE resource_id = ? ORDER BY checked_at DESC LIMIT 12 `).all(resource.id); stats.recentChecks = recentChecks.reverse(); pendingAlerts.push({ resource, incident, stats }); } // Slow alert: prefer explicit threshold, fall back to 2× P95 baseline const effectiveThreshold = resource.response_time_threshold || (resource.response_time_baseline ? resource.response_time_baseline * 2 : null); if (result.status === 'up' && effectiveThreshold && !resource.maintenance_mode) { if (result.response_time > effectiveThreshold) { const consecutive = (this.slowConsecutiveCount.get(resource.id) || 0) + 1; this.slowConsecutiveCount.set(resource.id, consecutive); const slowAlertConsecutiveSetting = db.prepare(`SELECT value FROM settings WHERE key = 'slow_alert_consecutive'`).get(); const slowAlertConsecutive = parseInt(slowAlertConsecutiveSetting?.value) || 3; if (consecutive >= slowAlertConsecutive) { const lastSlowAlert = this.slowAlertCooldown.get(resource.id) || 0; if (now - lastSlowAlert > 60 * 60 * 1000) { this.slowAlertCooldown.set(resource.id, now); await notificationService.sendAlert(resource, { type: 'slow', responseTime: result.response_time, threshold: effectiveThreshold, id: null, }); } } } else { // Reset consecutive counter when response time is back to normal this.slowConsecutiveCount.set(resource.id, 0); } } else if (!effectiveThreshold) { this.slowConsecutiveCount.set(resource.id, 0); } } catch (error) { // Check error handled silently } })()); await Promise.allSettled(tasks); // Flush collected alerts — grouped if multiple fired in the same cycle await notificationService.flushAlerts(pendingAlerts); // Broadcast updated dashboard to all connected WebSocket clients if (global.broadcastDashboardUpdate) { global.broadcastDashboardUpdate(); } } stop() { this.isRunning = false; } } module.exports = new Scheduler();