SenMeshNet/electron/bridges/inspection/inspectionScheduler.cjs

"use strict";

/**
 * 巡检调度引擎
 *
 * 职责：
 *   1. 定时扫描启用了巡检的设备
 *   2. 根据各设备的巡检周期，判断是否需要执行
 *   3. 并发执行多台设备的巡检
 *   4. 采集指标 → 判断告警 → 写入结果和告警记录
 *
 * 设计说明：
 *   - 每 30 秒轮询一次（检查哪些设备到巡检时间了）
 *   - 并发数默认 5，可配置
 *   - 单设备超时默认 30 秒
 *   - 巡检结果通过回调通知上层（更新 UI 用）
 */

const deviceStore = require("./deviceStore.cjs");
const alertRuleStore = require("./alertRuleStore.cjs");
const inspectionResultStore = require("./inspectionResultStore.cjs");
const alertRecordStore = require("./alertRecordStore.cjs");
const adapterRegistry = require("./adapterRegistry.cjs");
const { INSPECTION_STATUS } = require("./types.cjs");

const DEFAULT_POLL_INTERVAL_MS = 30 * 1000; // 30 秒轮询一次
const DEFAULT_CONCURRENCY = 5;
const DEFAULT_DEVICE_TIMEOUT_MS = 30 * 1000;

class InspectionScheduler {
  /**
   * @param {object} options
   * @param {function} options.readFn - 读取存储 (key) => string
   * @param {function} options.writeFn - 写入存储 (key, value) => void
   * @param {function} options.getConnectionFn - 获取 SSH 连接 (hostId) => connection 或 null
   * @param {function} options.execFn - 执行单条命令 (hostId, command) => Promise<string>
   * @param {function} [options.execCommandsFn] - 批量执行多条命令 (hostId, commands) => Promise<{ results, stdout }>
   * @param {function} [options.onResult] - 单设备巡检完成回调
   * @param {number} [options.pollIntervalMs] - 轮询间隔
   * @param {number} [options.concurrency] - 并发数
   * @param {number} [options.timeoutMs] - 单设备超时
   */
  constructor(options) {
    this.readFn = options.readFn;
    this.writeFn = options.writeFn;
    this.getConnectionFn = options.getConnectionFn;
    this.execFn = options.execFn;
    this.execCommandsFn = options.execCommandsFn;
    this.onResult = options.onResult || (() => {});
    this.onAlertChanged = options.onAlertChanged || (() => {});
    this.pollIntervalMs = options.pollIntervalMs || DEFAULT_POLL_INTERVAL_MS;
    this.concurrency = options.concurrency || DEFAULT_CONCURRENCY;
    this.timeoutMs = options.timeoutMs || DEFAULT_DEVICE_TIMEOUT_MS;
    // hostId → { hostname, label } 映射，用于告警记录显示设备名称
    this._hostInfo = options.hostInfo || {};

    this._timer = null;
    this._running = false;
    this._scanQueue = [];       // 待巡检的设备队列
    this._activeScans = 0;     // 当前正在执行的巡检数
  }

  // ---------- 生命周期 ----------

  start() {
    if (this._running) return;
    this._running = true;
    console.log("[InspectionScheduler] 巡检调度器已启动");

    // 启动时立即跑一轮（异步，不阻塞 start 返回）
    setImmediate(() => this._tick());
  }

  stop() {
    this._running = false;
    if (this._timer) {
      clearTimeout(this._timer);
      this._timer = null;
    }
    console.log("[InspectionScheduler] 巡检调度器已停止");
  }

  isRunning() {
    return this._running;
  }

  _scheduleNextTick() {
    if (!this._running) return;
    this._timer = setTimeout(() => this._tick(), this.pollIntervalMs);
  }

  // ---------- 主循环 ----------

  async _tick() {
    if (!this._running) return;

    try {
      // 全局自动巡检开关（一键启用/停止）
      const autoEnabledRaw = this.readFn("inspection-auto-enabled");
      const autoEnabled = autoEnabledRaw !== "false"; // 默认 true
      if (!autoEnabled) {
        return; // 全局停止 —— 跳过本次 tick
      }

      // 1. 获取所有启用巡检的设备
      const devices = deviceStore.getEnabledDevices(this.readFn);

      // 2. 筛选出到期需要巡检的设备
      const now = Date.now();
      const dueDevices = devices.filter((dev) => {
        const intervalMs = dev.effective_interval * 60 * 1000;
        const lastScan = dev.last_scan_at || 0;
        return now - lastScan >= intervalMs;
      });

      if (dueDevices.length > 0) {
        console.log(`[InspectionScheduler] 到期设备 ${dueDevices.length} 台，加入巡检队列`);
        // 加入队列
        for (const dev of dueDevices) {
          if (!this._scanQueue.find((d) => d.host_id === dev.host_id)) {
            this._scanQueue.push(dev);
          }
        }
        // 启动并发执行
        this._processQueue();
      }
    } catch (e) {
      console.error("[InspectionScheduler] 调度异常:", e.message);
    }

    // 3. 安排下一次轮询
    this._scheduleNextTick();
  }

  _processQueue() {
    while (this._activeScans < this.concurrency && this._scanQueue.length > 0) {
      const device = this._scanQueue.shift();
      this._activeScans++;
      this._scanDevice(device).finally(() => {
        this._activeScans--;
        // 继续处理队列
        this._processQueue();
      });
    }
  }

  // ---------- 单设备巡检 ----------

  async _scanDevice(deviceExt) {
    const startTime = Date.now();
    const hostId = deviceExt.host_id;

    try {
      // 1. 获取设备类型信息
      const catalog = deviceStore.getCatalogById(deviceExt.catalog_id);
      if (!catalog) {
        throw new Error(`未找到设备类型: ${deviceExt.catalog_id}`);
      }

      // 2. 创建采集器
      // connectionInfo 只传 { host: hostId } 作为标识——adapter 的 execCommands 会被 options.execCommands 覆盖，
      // 真正的 SSH 凭据由主进程 execCommandsFn 从 _hostConfigs 读取（和 execFn 同理）
      const adapter = adapterRegistry.createAdapter(
        catalog.adapter_class,
        { host: hostId },
        {
          execFn: async (cmd) => {
            if (!this.execFn) throw new Error("execFn 未配置");
            return this.execFn(hostId, cmd);
          },
          execCommands: this.execCommandsFn
            ? async (commands, options) => {
                return this.execCommandsFn(hostId, commands, options);
              }
            : undefined,
          timeoutMs: this.timeoutMs,
        }
      );

      if (!adapter) {
        throw new Error(`采集器未注册: ${catalog.adapter_class}`);
      }

      // 3. 执行采集（带超时）
      const result = await this._withTimeout(
        adapter.collect(),
        this.timeoutMs,
        "巡检超时"
      );

      const durationMs = Date.now() - startTime;

      // 4. 保存巡检结果（带上 catalog_id 供后续 RCA 根因分析用）
      const record = inspectionResultStore.addResult(
        this.readFn,
        this.writeFn,
        {
          device_id: hostId,
          catalog_id: deviceExt.catalog_id || catalog.id || null,
          online: result.online,
          status: result.status,
          duration_ms: durationMs,
          metrics: result.metrics,
          alerts: [],
          error_message: result.error,
        }
      );

      // 5. 更新设备最后巡检状态
      deviceStore.updateLastScan(
        this.readFn,
        this.writeFn,
        hostId,
        result.status,
        result.error
      );

      // 6. 判断告警
      const triggeredAlerts = this._evaluateAlerts(deviceExt, catalog, result.metrics);
      record.alerts = triggeredAlerts;

      if (triggeredAlerts.length > 0) this.onAlertChanged();

      // 7. 回调通知
      this.onResult({
        device_id: hostId,
        result: record,
        alerts: triggeredAlerts,
      });

      return record;
    } catch (e) {
      const durationMs = Date.now() - startTime;

      // 失败也要记录
      const record = inspectionResultStore.addResult(
        this.readFn,
        this.writeFn,
        {
          device_id: hostId,
          online: false,
          status: e.message.includes("超时")
            ? INSPECTION_STATUS.TIMEOUT
            : INSPECTION_STATUS.FAILED,
          duration_ms: durationMs,
          metrics: {},
          alerts: [],
          error_message: e.message,
        }
      );

      deviceStore.updateLastScan(
        this.readFn,
        this.writeFn,
        hostId,
        e.message.includes("超时") ? INSPECTION_STATUS.TIMEOUT : INSPECTION_STATUS.FAILED,
        e.message
      );

      // 执行错误告警 —— 统一归类为 rule_offline（与设备离线告警合并）
      // 不再额外调 _evaluateAlerts(online: false) 避免重复创建同规则告警
      const failStatus = e.message.includes("超时") ? "timeout" : "failed";
      alertRecordStore.createExecutionErrorAlert(
        this.readFn,
        this.writeFn,
        {
          device_id: hostId,
          device_name: deviceExt.label || deviceExt.hostname || hostId,
          catalog_id: deviceExt.catalog_id,
        },
        failStatus,
        e.message || "巡检执行失败"
      );
      this.onAlertChanged();

      this.onResult({
        device_id: hostId,
        result: record,
        alerts: [],
        error: e.message,
      });

      return record;
    }
  }

  // ---------- 告警判断 ----------

  _evaluateAlerts(deviceExt, catalog, metrics) {
    const triggered = [];

    try {
      // 获取该设备适用的规则
      const rules = alertRuleStore.getRulesForDevice(this.readFn, deviceExt);
      const hostInfo = this._hostInfo[deviceExt.host_id] || {};
      const deviceName = hostInfo.label || hostInfo.hostname || deviceExt.host_id;
      console.log(`[InspectionScheduler] 告警评估 device=${deviceName}(${deviceExt.host_id}) online=${metrics.online} 规则数=${rules.length}`);

      for (const rule of rules) {
        if (rule.metric === "online") {
          const isOnline = metrics.online !== false;
          const shouldAlert = rule.threshold === "false" ? !isOnline : isOnline;
          console.log(`[InspectionScheduler] 离线规则: name=${rule.name} enabled=${rule.enabled} threshold=${rule.threshold} isOnline=${isOnline} shouldAlert=${shouldAlert}`);
          const alertResult = alertRecordStore.processAlert(
            this.readFn,
            this.writeFn,
            { device_id: deviceExt.host_id, device_name: deviceName },
            rule,
            isOnline ? "在线" : "离线",
            shouldAlert
          );
          if (alertResult) {
            console.log(`[InspectionScheduler] 告警触发:`, JSON.stringify(alertResult));
            triggered.push(alertResult);
          }
          continue;
        }

        // 普通数值指标
        const value = this._getMetricValue(metrics, rule.metric);
        if (value === null || value === undefined) continue;

        const isTriggered = alertRuleStore.evaluateRule(value, rule.operator, rule.threshold);

        // 注意：hostInfo / deviceName 在函数顶部 L285-286 已声明，此处直接复用，
        // 避免同名 const 重复声明导致 TDZ（Cannot access 'deviceName' before initialization）
        const alertResult = alertRecordStore.processAlert(
          this.readFn,
          this.writeFn,
          { device_id: deviceExt.host_id, device_name: deviceName },
          rule,
          value,
          isTriggered
        );

        if (alertResult) triggered.push(alertResult);
      }
    } catch (e) {
      console.error("[InspectionScheduler] 告警判断异常:", e.message);
    }

    return triggered;
  }

  _getMetricValue(metrics, metricKey) {
    if (!metrics) return null;
    // 直接取
    if (metrics[metricKey] !== undefined) return metrics[metricKey];
    // 磁盘最高使用率
    if (metricKey === "disk_usage" && Array.isArray(metrics.disk_usage)) {
      return Math.max(...metrics.disk_usage.map((d) => d.usage_percent || 0));
    }
    return null;
  }

  // ---------- 手动巡检 ----------

  /**
   * 手动触发单设备巡检
   */
  async scanDeviceNow(hostId) {
    const ext = deviceStore.getExtension(this.readFn, hostId);
    if (!ext) throw new Error(`设备未配置巡检: ${hostId}`);

    const interval = deviceStore.getEffectiveInterval(this.readFn, hostId);
    const device = { ...ext, effective_interval: interval };

    // 加入队列头部
    this._scanQueue.unshift(device);
    this._processQueue();

    return { queued: true, host_id: hostId };
  }

  /**
   * 手动触发所有启用巡检的设备
   */
  async scanAllNow() {
    const devices = deviceStore.getEnabledDevices(this.readFn);
    let count = 0;
    for (const dev of devices) {
      if (!this._scanQueue.find((d) => d.host_id === dev.host_id)) {
        this._scanQueue.push(dev);
        count++;
      }
    }
    this._processQueue();
    return { queued: count };
  }

  // ---------- 工具 ----------

  _withTimeout(promise, ms, message) {
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => reject(new Error(message || "超时")), ms);
      promise.then(
        (v) => { clearTimeout(timer); resolve(v); },
        (e) => { clearTimeout(timer); reject(e); }
      );
    });
  }

  getStatus() {
    return {
      running: this._running,
      queue_size: this._scanQueue.length,
      active_scans: this._activeScans,
      concurrency: this.concurrency,
    };
  }
}

module.exports = InspectionScheduler;