skywatch/server/scheduler.js

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();