DiAL/src/server/checker/engine.ts

import { groupBy, isError, Semaphore } from "es-toolkit";

import type { ProbeStore } from "./store";
import type { CheckResult, ResolvedTargetBase } from "./types";

import { errorFailure } from "./expect/failure";
import { checkerRegistry } from "./runner";

const PRUNE_INTERVAL_MS = 3600000;

export class ProbeEngine {
  private retentionMs: number;
  private semaphore: Semaphore;
  private store: ProbeStore;
  private targetNameToId = new Map<string, number>();
  private targets: ResolvedTargetBase[];
  private timers: Array<ReturnType<typeof setInterval>> = [];

  constructor(store: ProbeStore, targets: ResolvedTargetBase[], maxConcurrentChecks?: number, retentionMs?: number) {
    this.store = store;
    this.targets = targets;
    this.semaphore = new Semaphore(maxConcurrentChecks ?? 20);
    this.retentionMs = retentionMs ?? 0;
    this.refreshCache();
  }

  start(): void {
    const groups = groupBy(this.targets, (t) => t.intervalMs);

    for (const [intervalMs, groupTargets] of Object.entries(groups)) {
      void this.probeGroup(groupTargets);

      const timer = setInterval(() => {
        void this.probeGroup(groupTargets);
      }, Number(intervalMs));

      this.timers.push(timer);
    }

    if (this.retentionMs > 0) {
      this.store.prune(this.retentionMs);
      const pruneTimer = setInterval(() => {
        this.store.prune(this.retentionMs);
      }, PRUNE_INTERVAL_MS);
      this.timers.push(pruneTimer);
    }
  }

  stop(): void {
    for (const timer of this.timers) {
      clearInterval(timer);
    }
    this.timers = [];
  }

  private async probeGroup(targets: ResolvedTargetBase[]): Promise<void> {
    const results = await Promise.allSettled(
      targets.map(async (target) => {
        await this.semaphore.acquire();
        try {
          return await this.runCheck(target);
        } finally {
          this.semaphore.release();
        }
      }),
    );

    for (const [index, result] of results.entries()) {
      if (result.status === "fulfilled") {
        this.writeResult(result.value);
      } else {
        const target = targets[index];
        console.warn("探针执行失败:", result.reason);
        if (!target) continue;
        this.writeResult({
          durationMs: null,
          failure: errorFailure("internal", "engine", formatReason(result.reason)),
          matched: false,
          statusDetail: null,
          targetName: target.name,
          timestamp: new Date().toISOString(),
        });
      }
    }
  }

  private refreshCache(): void {
    this.targetNameToId.clear();
    for (const target of this.store.getTargets()) {
      this.targetNameToId.set(target.name, target.id);
    }
  }

  private async runCheck(target: ResolvedTargetBase): Promise<CheckResult> {
    const checker = checkerRegistry.get(target.type);
    const controller = new AbortController();
    const timeoutId = setTimeout(() => controller.abort(), target.timeoutMs);

    try {
      return await checker.execute(target, { signal: controller.signal });
    } finally {
      clearTimeout(timeoutId);
    }
  }

  private writeResult(result: CheckResult): void {
    const targetId = this.targetNameToId.get(result.targetName);
    if (!targetId) return;

    this.store.insertCheckResult({
      durationMs: result.durationMs,
      failure: result.failure,
      matched: result.matched,
      statusDetail: result.statusDetail,
      targetId,
      timestamp: result.timestamp,
    });
  }
}

function formatReason(reason: unknown): string {
  return isError(reason) ? reason.message : String(reason);
}