skywatch/server/monitorService.js

const axios = require('axios');
const http = require('http');
const https = require('https');
const net = require('net');
const tls = require('tls');
const dns = require('dns').promises;
const WebSocket = require('ws');
const ping = require('ping');
const db = require('./database');
const RetryPolicy = require('./retryPolicy');
const CircuitBreaker = require('./circuitBreaker');

const DEFAULT_PORTS = {
  http: 80,
  https: 443,
  tcp: 80,
  tls: 443,
  websocket: 80,
};

function parseHostPort(urlString, fallbackPort) {
  try {
    const u = new URL(urlString.startsWith('http') || urlString.startsWith('ws') ? urlString : `http://${urlString}`);
    return { host: u.hostname, port: u.port ? Number(u.port) : fallbackPort };
  } catch (err) {
    // Fallback if url is just host:port
    const [host, port] = urlString.split(':');
    return { host, port: port ? Number(port) : fallbackPort };
  }
}

class MonitorService {
  constructor() {
    // Disable keep-alive to avoid "Parse Error: Data after Connection: close"
    // crashes. Keep-alive connection reuse causes the HTTP parser to throw
    // when a monitored server sends extra data after Connection: close — the
    // socket-level error becomes an uncaught exception that kills the process.
    // Monitoring checks are infrequent enough that connection pooling provides
    // no meaningful benefit.
    this.httpAgent = new http.Agent({ keepAlive: false });
    this.httpsAgent = new https.Agent({ keepAlive: false });
    
    // Circuit breaker for HTTP checks (5 failures before opening, 60s reset)
    this.httpCircuitBreaker = new CircuitBreaker(5, 60000, this.isTransientError);
    
    // Retry policy: 3 retries with exponential backoff
    this.retryPolicy = new RetryPolicy(3, 1000, 10000);
    
    // Circuit breakers per resource for fine-grained control
    this.resourceCircuitBreakers = new Map();
  }

  getResourceCircuitBreaker(resourceId) {
    if (!this.resourceCircuitBreakers.has(resourceId)) {
      this.resourceCircuitBreakers.set(resourceId, new CircuitBreaker(3, 120000, this.isTransientError));
    }
    return this.resourceCircuitBreakers.get(resourceId);
  }

  isTransientError(error) {
    // Only retry on transient errors (network, timeout, 5xx)
    if (error.code === 'ECONNREFUSED' || error.code === 'ECONNRESET' || error.code === 'ETIMEDOUT') {
      return true;
    }
    if (error.response?.status >= 500) {
      return true;
    }
    return false;
  }
  async checkResource(resource) {
    const startTime = Date.now();
    const base = {
      resource_id: resource.id,
      status: 'down',
      response_time: null,
      status_code: null,
      error_message: null,
      details: null,
    };

    // Check if this is a transaction
    if (resource.is_transaction) {
      return this.checkTransaction(resource, startTime, base);
    }

    const type = (resource.type || 'http').toLowerCase();

    switch (type) {
      case 'http':
      case 'https':
      case 'health':
        return this.checkHttp(resource, startTime, base);
      case 'tcp':
        return this.checkTcp(resource, startTime, base);
      case 'tls':
        return this.checkTls(resource, startTime, base);
      case 'dns':
        return this.checkDns(resource, startTime, base);
      case 'websocket':
      case 'ws':
      case 'wss':
        return this.checkWebSocket(resource, startTime, base);
      case 'icmp':
      case 'ping':
        return this.checkIcmp(resource, startTime, base);
      case 'heartbeat':
        return this.checkHeartbeat(resource, startTime, base);
      default:
        // Fallback to HTTP if unknown
        return this.checkHttp(resource, startTime, base);
    }
  }

  checkHeartbeat(resource, startTime, base) {
    const now = Date.now();
    const lastPing = resource.last_heartbeat_at
      ? new Date(resource.last_heartbeat_at).getTime()
      : null;
    const timeout = resource.heartbeat_timeout || 300000;

    if (!lastPing) {
      return { ...base, status: 'down', response_time: null, error_message: 'No heartbeat received yet' };
    }
    const elapsed = now - lastPing;
    if (elapsed > timeout) {
      return {
        ...base,
        status: 'down',
        response_time: elapsed,
        error_message: `Missed heartbeat — ${Math.round(elapsed / 1000)}s since last ping (timeout: ${timeout / 1000}s)`,
      };
    }
    return { ...base, status: 'up', response_time: elapsed };
  }

  async checkHttp(resource, startTime, base) {
    let result = { ...base };
    const circuitBreaker = this.getResourceCircuitBreaker(resource.id);
    
    try {
      await circuitBreaker.execute(async () => {
        return await this.retryPolicy.execute(async () => {
          const headers = {};
          if (resource.http_headers) {
            try {
              const customHeaders = JSON.parse(resource.http_headers);
              Object.assign(headers, customHeaders);
            } catch (e) {
              // Invalid JSON, skip custom headers
            }
          }

          const method = (resource.http_method || 'GET').toUpperCase();
          const requestConfig = {
            method,
            url: resource.url,
            timeout: resource.timeout,
            validateStatus: () => true,
            maxRedirects: 5,
            headers,
            httpAgent: this.httpAgent,
            httpsAgent: this.httpsAgent,
          };
          if (resource.http_body && ['POST', 'PUT', 'PATCH'].includes(method)) {
            try {
              requestConfig.data = JSON.parse(resource.http_body);
              if (!headers['Content-Type']) headers['Content-Type'] = 'application/json';
            } catch {
              requestConfig.data = resource.http_body;
            }
          }
          const response = await axios(requestConfig);
          const responseTime = Date.now() - startTime;
          const isUp = response.status >= 200 && response.status < 400;
          
          // Check for keyword if specified
          let keywordMatch = true;
          if (resource.http_keyword && response.data) {
            const dataStr = typeof response.data === 'string' ? response.data : JSON.stringify(response.data);
            keywordMatch = dataStr.includes(resource.http_keyword);
            if (!keywordMatch) {
              result.error_message = `Keyword "${resource.http_keyword}" not found in response`;
            }
          }

          result = {
            ...result,
            status: (isUp && keywordMatch) ? 'up' : 'down',
            response_time: responseTime,
            status_code: response.status,
            error_message: isUp && keywordMatch ? null : (result.error_message || `HTTP ${response.status}`),
          };
          
          return result;
        }, `HTTP check for ${resource.name}`);
      });
    } catch (error) {
      const responseTime = Date.now() - startTime;
      result.error_message = error.originalError?.message || error.message;
      result.response_time = responseTime;
      result.status = 'down';
    }
    return result;
  }

  async checkTcp(resource, startTime, base) {
    const { host, port } = parseHostPort(resource.url, DEFAULT_PORTS.tcp);
    return new Promise(resolve => {
      const socket = new net.Socket();
      let finished = false;

      const done = (status, error_message = null) => {
        if (finished) return;
        finished = true;
        socket.destroy();
        resolve({
          ...base,
          status,
          response_time: Date.now() - startTime,
          status_code: null,
          error_message,
          details: JSON.stringify({ host, port }),
        });
      };

      socket.setTimeout(resource.timeout || 5000);
      socket.once('connect', () => done('up'));
      socket.once('timeout', () => done('down', 'TCP timeout'));
      socket.once('error', (err) => done('down', err.message));
      socket.connect(port, host);
    });
  }

  async checkTls(resource, startTime, base) {
    const { host, port } = parseHostPort(resource.url, DEFAULT_PORTS.tls);
    return new Promise(resolve => {
      const socket = tls.connect({ host, port, servername: host, timeout: resource.timeout || 5000 }, () => {
        const cert = socket.getPeerCertificate();
        const now = Date.now();
        const expires = new Date(cert.valid_to).getTime();
        const daysRemaining = Math.round((expires - now) / (1000 * 60 * 60 * 24));
        const validHost = cert.subjectaltname ? cert.subjectaltname.includes(host) : true;
        
        // Check certificate expiry threshold
        const expiryThreshold = resource.cert_expiry_days || 30;
        const certExpiringWarning = daysRemaining <= expiryThreshold && daysRemaining > 0;

        const status = daysRemaining > 0 && validHost ? 'up' : 'down';
        resolve({
          ...base,
          status,
          response_time: Date.now() - startTime,
          error_message: status === 'up' ? null : 'TLS invalid or expired',
          cert_expiry_date: cert.valid_to,
          details: JSON.stringify({
            issuer: cert.issuer?.CN,
            subject: cert.subject?.CN,
            valid_to: cert.valid_to,
            days_remaining: daysRemaining,
            host_matches: validHost,
            cert_expiring_soon: certExpiringWarning,
            expiry_threshold: expiryThreshold,
          }),
        });
        socket.end();
      });

      socket.on('error', err => {
        resolve({
          ...base,
          status: 'down',
          response_time: Date.now() - startTime,
          error_message: err.message,
        });
      });

      socket.setTimeout(resource.timeout || 5000, () => {
        resolve({
          ...base,
          status: 'down',
          response_time: Date.now() - startTime,
          error_message: 'TLS timeout',
        });
        socket.destroy();
      });
    });
  }

  async checkDns(resource, startTime, base) {
    const { host } = parseHostPort(resource.url, DEFAULT_PORTS.http);
    try {
      const records = await dns.lookup(host, { all: true });
      return {
        ...base,
        status: 'up',
        response_time: Date.now() - startTime,
        details: JSON.stringify({ answers: records }),
      };
    } catch (err) {
      return {
        ...base,
        status: 'down',
        response_time: Date.now() - startTime,
        error_message: err.message,
      };
    }
  }

  async checkWebSocket(resource, startTime, base) {
    return new Promise(resolve => {
      let finished = false;

      const done = (status, error_message = null) => {
        if (finished) return;
        finished = true;
        resolve({
          ...base,
          status,
          response_time: Date.now() - startTime,
          error_message,
        });
      };

      try {
        const ws = new WebSocket(resource.url, { handshakeTimeout: resource.timeout || 5000 });
        ws.on('open', () => {
          done('up');
          ws.close();
        });
        ws.on('error', (err) => done('down', err.message));
        ws.on('close', () => done('up'));
      } catch (err) {
        done('down', err.message);
      }
    });
  }

  async checkIcmp(resource, startTime, base) {
    const { host } = parseHostPort(resource.url, DEFAULT_PORTS.http);
    try {
      const res = await ping.promise.probe(host, { timeout: (resource.timeout || 5000) / 1000 });
      const status = res.alive ? 'up' : 'down';
      return {
        ...base,
        status,
        response_time: res.time || Date.now() - startTime,
        error_message: res.alive ? null : res.output,
        details: JSON.stringify({ packetLoss: res.packetLoss, time: res.time }),
      };
    } catch (err) {
      return {
        ...base,
        status: 'down',
        response_time: Date.now() - startTime,
        error_message: err.message,
      };
    }
  }

  saveCheck(check) {
    const stmt = db.prepare(`
      INSERT INTO checks (resource_id, status, response_time, status_code, error_message, details, checked_at)
      VALUES (?, ?, ?, ?, ?, ?, datetime('now'))
    `);
    stmt.run(
      check.resource_id,
      check.status,
      check.response_time,
      check.status_code,
      check.error_message,
      check.details || null
    );
  }

  getLastCheck(resourceId) {
    return db.prepare(`
      SELECT 
        id,
        resource_id,
        status,
        response_time,
        status_code,
        error_message,
        details,
        REPLACE(checked_at, ' ', 'T') || 'Z' AS checked_at
      FROM checks 
      WHERE resource_id = ? 
      ORDER BY checked_at DESC 
      LIMIT 1
    `).get(resourceId);
  }

  handleIncident(resourceId, isDown) {
    const resource = db.prepare('SELECT * FROM resources WHERE id = ?').get(resourceId);
    
    // Per-resource threshold overrides global setting
    const setting = db.prepare("SELECT value FROM settings WHERE key = 'incident_failure_threshold'").get();
    const globalThreshold = parseInt(setting?.value || '10');
    const threshold = (resource?.consecutive_failures_threshold > 0)
      ? resource.consecutive_failures_threshold
      : globalThreshold;
    
    const activeIncident = db.prepare(`
      SELECT * FROM incidents 
      WHERE resource_id = ? AND resolved_at IS NULL
    `).get(resourceId);

    if (isDown) {
      // Check consecutive failures
      const recentChecks = db.prepare(`
        SELECT status, error_message, checked_at FROM checks
        WHERE resource_id = ?
        ORDER BY checked_at DESC
        LIMIT ?
      `).all(resourceId, threshold);

      const firstUpIdx = recentChecks.findIndex(c => c.status === 'up');
      const consecutiveFailures = firstUpIdx === -1 ? recentChecks.length : firstUpIdx;

      // Only trigger incident if we've hit the threshold
      if (consecutiveFailures >= threshold && !activeIncident) {
        // Build description with failure reason and first failed check timestamp
        const lastCheck = recentChecks[0];
        const firstFailedCheck = recentChecks[recentChecks.length - 1];
        
        const reason = lastCheck?.error_message || 'Service returned down status';
        const description = `Resource down after ${consecutiveFailures} consecutive failed checks.\nReason: ${reason}\nFirst failure: ${firstFailedCheck?.checked_at}`;

        const stmt = db.prepare(`
          INSERT INTO incidents (resource_id, description, failed_check_count) VALUES (?, ?, ?)
        `);
        const result = stmt.run(resourceId, description, consecutiveFailures);
        return { 
          type: 'started', 
          incident: true,
          consecutiveFailures,
          description
        };
      }
    } else if (!isDown && activeIncident) {
      // Resolve incident
      db.prepare(`
        UPDATE incidents 
        SET resolved_at = CURRENT_TIMESTAMP 
        WHERE id = ?
      `).run(activeIncident.id);
      return { type: 'resolved', incident: activeIncident };
    }

    return { type: 'none' };
  }

  getResourceStats(resourceId, hours = 24) {
    const checks = db.prepare(`
      SELECT 
        id,
        resource_id,
        status,
        response_time,
        status_code,
        error_message,
        details,
        REPLACE(checked_at, ' ', 'T') || 'Z' AS checked_at
      FROM checks 
      WHERE resource_id = ? 
      AND checked_at >= datetime('now', '-' || ? || ' hours')
      ORDER BY checked_at ASC
    `).all(resourceId, hours);

    const upChecks = checks.filter(c => c.status === 'up');
    const totalChecks = checks.length;
    const uptime = totalChecks > 0 ? (upChecks.length / totalChecks) * 100 : 0;

    const responseTimes = checks
      .filter(c => c.response_time !== null)
      .map(c => c.response_time);

    const avgResponseTime = responseTimes.length > 0
      ? responseTimes.reduce((a, b) => a + b, 0) / responseTimes.length
      : 0;

    return {
      uptime: uptime.toFixed(2),
      avgResponseTime: avgResponseTime.toFixed(0),
      totalChecks,
      checks: checks.slice(-50), // Last 50 checks
    };
  }

  /**
   * Paginated checks with basic filters and ISO timestamps
   */
  getChecks(resourceId, {
    limit = 50,
    offset = 0,
    status = null,
    from = null,
    to = null,
    sort = 'desc',
  } = {}) {
    const clauses = ['resource_id = ?'];
    const params = [resourceId];

    if (status) {
      clauses.push('status = ?');
      params.push(status);
    }
    if (from) {
      clauses.push('checked_at >= ?');
      params.push(from);
    }
    if (to) {
      clauses.push('checked_at <= ?');
      params.push(to);
    }

    const order = sort === 'asc' ? 'ASC' : 'DESC';

    const sql = `
      SELECT
        id,
        resource_id,
        status,
        response_time,
        status_code,
        error_message,
        details,
        REPLACE(checked_at, ' ', 'T') || 'Z' AS checked_at
      FROM checks
      WHERE ${clauses.join(' AND ')}
      ORDER BY checked_at ${order}
      LIMIT ? OFFSET ?
    `;
    return db.prepare(sql).all(...params, limit, offset);
  }

  /**
   * Incidents timeline with filters
   */
  getIncidents(resourceId, {
    limit = 50,
    offset = 0,
    status = 'all', // all | open | closed
    from = null,
    to = null,
    sort = 'desc',
  } = {}) {
    const clauses = ['resource_id = ?'];
    const params = [resourceId];

    if (status === 'open') {
      clauses.push('resolved_at IS NULL');
    } else if (status === 'closed') {
      clauses.push('resolved_at IS NOT NULL');
    }

    if (from) {
      clauses.push('started_at >= ?');
      params.push(from);
    }
    if (to) {
      clauses.push('started_at <= ?');
      params.push(to);
    }

    const order = sort === 'asc' ? 'ASC' : 'DESC';
    const sql = `
      SELECT
        id,
        resource_id,
        REPLACE(started_at, ' ', 'T') || 'Z' AS started_at,
        CASE
          WHEN resolved_at IS NULL THEN NULL
          ELSE REPLACE(resolved_at, ' ', 'T') || 'Z'
        END AS resolved_at,
        description,
        failed_check_count
      FROM incidents
      WHERE ${clauses.join(' AND ')}
      ORDER BY started_at ${order}
      LIMIT ? OFFSET ?
    `;
    return db.prepare(sql).all(...params, limit, offset);
  }

  /**
   * SLA/SLO summary over a time window
   */
  getSlaSummary(resourceId, hours = 24) {
    const checks = db.prepare(`
      SELECT
        status,
        response_time,
        REPLACE(checked_at, ' ', 'T') || 'Z' AS checked_at
      FROM checks
      WHERE resource_id = ?
        AND checked_at >= datetime('now', '-' || ? || ' hours')
      ORDER BY checked_at ASC
    `).all(resourceId, hours);

    if (checks.length < 2) {
      return {
        windowHours: hours,
        uptimePct: 0,
        downtimeMinutes: 0,
        mttrMinutes: null,
        mtbfMinutes: null,
        p95LatencyMs: null,
        totalChecks: checks.length,
      };
    }

    let upMs = 0;
    let downMs = 0;
    const upRuns = [];
    const downRuns = [];

    for (let i = 0; i < checks.length - 1; i++) {
      const cur = checks[i];
      const next = checks[i + 1];
      const curTime = new Date(cur.checked_at).getTime();
      const nextTime = new Date(next.checked_at).getTime();
      const span = Math.max(0, nextTime - curTime);

      if (cur.status === 'up') {
        upMs += span;
        upRuns.push(span);
      } else {
        downMs += span;
        downRuns.push(span);
      }
    }

    const totalMs = upMs + downMs || 1;
    const uptimePct = 100 * (upMs / totalMs);

    const avg = arr => arr.length ? arr.reduce((a, b) => a + b, 0) / arr.length : null;
    const mttrMs = avg(downRuns);
    const mtbfMs = avg(upRuns);

    const latencies = checks
      .map(c => c.response_time)
      .filter(v => typeof v === 'number')
      .sort((a, b) => a - b);
    const p95LatencyMs = latencies.length
      ? latencies[Math.floor(0.95 * (latencies.length - 1))]
      : null;

    return {
      windowHours: hours,
      uptimePct: Number(uptimePct.toFixed(2)),
      downtimeMinutes: Number((downMs / 60000).toFixed(2)),
      mttrMinutes: mttrMs != null ? Number((mttrMs / 60000).toFixed(2)) : null,
      mtbfMinutes: mtbfMs != null ? Number((mtbfMs / 60000).toFixed(2)) : null,
      p95LatencyMs,
      totalChecks: checks.length,
    };
  }

  // Check multi-step transaction
  async checkTransaction(resource, startTime, base) {
    try {
      const steps = db.prepare(`
        SELECT * FROM transaction_checks
        WHERE resource_id = ?
        ORDER BY step_order ASC
      `).all(resource.id);

      if (steps.length === 0) {
        return {
          ...base,
          status: 'down',
          error_message: 'No transaction steps defined'
        };
      }

      let totalTime = 0;
      const results = [];

      for (let i = 0; i < steps.length; i++) {
        const step = steps[i];
        const stepStart = Date.now();

        try {
          const headers = {};
          if (step.headers) {
            try {
              Object.assign(headers, JSON.parse(step.headers));
            } catch (e) {
              // Invalid JSON, skip
            }
          }

          const body = step.body ? JSON.parse(step.body) : undefined;
          const response = await axios({
            method: step.method || 'GET',
            url: step.url,
            headers,
            data: body,
            timeout: resource.timeout,
            validateStatus: () => true,
            maxRedirects: 5,
          });

          const stepTime = Date.now() - stepStart;
          const statusOk = response.status === (step.expected_status || 200);
          
          let keywordMatch = true;
          if (step.keyword && response.data) {
            const dataStr = typeof response.data === 'string' ? response.data : JSON.stringify(response.data);
            keywordMatch = dataStr.includes(step.keyword);
          }

          if (!statusOk || !keywordMatch) {
            return {
              ...base,
              status: 'down',
              response_time: Date.now() - startTime,
              status_code: response.status,
              error_message: `Transaction failed at step ${i + 1}: Expected ${step.expected_status || 200}, got ${response.status}${!keywordMatch ? ' (keyword not found)' : ''}`
            };
          }

          results.push({
            step: i + 1,
            status: response.status,
            time: stepTime
          });

          totalTime += stepTime;
        } catch (error) {
          return {
            ...base,
            status: 'down',
            response_time: Date.now() - startTime,
            error_message: `Transaction failed at step ${i + 1}: ${error.message}`
          };
        }
      }

      // All steps passed
      const responseTime = Date.now() - startTime;
      return {
        ...base,
        status: 'up',
        response_time: responseTime,
        status_code: 200,
        details: JSON.stringify({ steps: steps.length, results })
      };
    } catch (error) {
      return {
        ...base,
        status: 'down',
        response_time: Date.now() - startTime,
        error_message: error.message
      };
    }
  }

  // Calculate baseline response time for anomaly detection
  calculateResponseTimeBaseline(resourceId, days = 7) {
    const checks = db.prepare(`
      SELECT response_time FROM checks
      WHERE resource_id = ? AND status = 'up' AND response_time IS NOT NULL
      AND checked_at > datetime('now', ?)
      ORDER BY response_time
    `).all(resourceId, `-${days} days`);

    if (checks.length === 0) {
      return null;
    }

    // Use P95 so the baseline reflects realistic worst-case, not the average.
    // Threshold is 2× P95, making it very unlikely to trigger on normal variance.
    const p95Index = Math.min(Math.floor(checks.length * 0.95), checks.length - 1);
    return Math.round(checks[p95Index].response_time);
  }

  // Check if response time is anomalous
  isResponseTimeAnomaly(resourceId, currentResponseTime, deviationPercent = 50) {
    try {
      const resource = db.prepare('SELECT * FROM resources WHERE id = ?').get(resourceId);
      if (!resource || !resource.response_time_baseline) {
        return false;
      }

      const baseline = resource.response_time_baseline;
      const threshold = (baseline * (100 + deviationPercent)) / 100;

      return currentResponseTime > threshold;
    } catch (error) {
      return false;
    }
  }
}

module.exports = new MonitorService();