MikrotikManager/backend/src/services/alert-engine/signals-from-collectors.ts

import { sqliteDatabase } from "../../db/index.js"
import type {
  GreBgpSnapshotRunSnapshot,
  PingRunSnapshot,
  ResourcesRunSnapshot,
  SchedulerRunSnapshot,
  ServersRestPingRunSnapshot,
  TrafficRunSnapshot,
} from "../../types/scheduler-run-snapshot.js"
import type {
  BgpPeerSignal,
  GreTunnelSignal,
  ProbeSignal,
  ServerSignal,
  SignalSnapshot,
  TrafficServerSignal,
} from "./types.js"

function normalizeNameKey(v: string): string {
  return v.trim().toLowerCase()
}

function safeParseRunSnapshot(raw: string | null): SchedulerRunSnapshot | null {
  if (!raw) return null
  try {
    return JSON.parse(raw) as SchedulerRunSnapshot
  } catch {
    return null
  }
}

function readLatestOkRunSnapshots(jobKey: string, limit = 32): SchedulerRunSnapshot[] {
  const rows = sqliteDatabase
    .prepare(
      `SELECT result_json AS resultJson
       FROM scheduler_runs
       WHERE job_key = ? AND status = 'ok' AND result_json IS NOT NULL
       ORDER BY finished_at DESC
       LIMIT ?`,
    )
    .all(jobKey, limit) as { resultJson: string | null }[]
  const out: SchedulerRunSnapshot[] = []
  for (const row of rows) {
    const parsed = safeParseRunSnapshot(row.resultJson)
    if (!parsed || parsed.job !== jobKey) continue
    out.push(parsed)
  }
  return out
}

type TriState = "online" | "offline" | "degraded"

function timelineToServerSignal(name: string, sampledAt: string, statuses: TriState[]): ServerSignal {
  return {
    name,
    status: statuses[0] ?? "offline",
    prevStatus: statuses[1] ?? null,
    prev2Status: statuses[2] ?? null,
    sampledAt,
  }
}

function collectServerSignalsFromResourceRuns(runs: ResourcesRunSnapshot[]): Map<string, ServerSignal> {
  const byKey = new Map<
    string,
    { name: string; sampledAt: string; statuses: TriState[] }
  >()
  for (const run of runs) {
    for (const s of run.servers ?? []) {
      const name = String(s.name ?? "").trim()
      if (!name) continue
      const key = normalizeNameKey(name)
      const rec = byKey.get(key) ?? { name, sampledAt: run.sampledAt, statuses: [] }
      if (rec.statuses.length < 3) rec.statuses.push(s.status === "online" ? "online" : "offline")
      if (!byKey.has(key)) rec.sampledAt = run.sampledAt
      byKey.set(key, rec)
    }
  }
  const out = new Map<string, ServerSignal>()
  for (const [key, rec] of byKey) out.set(key, timelineToServerSignal(rec.name, rec.sampledAt, rec.statuses))
  return out
}

function collectServerSignalsFromRestRuns(runs: ServersRestPingRunSnapshot[]): Map<string, ServerSignal> {
  const byKey = new Map<
    string,
    { name: string; sampledAt: string; statuses: TriState[] }
  >()
  for (const run of runs) {
    for (const s of run.servers ?? []) {
      const name = String(s.name ?? "").trim()
      if (!name) continue
      const key = normalizeNameKey(name)
      const rec = byKey.get(key) ?? { name, sampledAt: run.sampledAt, statuses: [] }
      if (rec.statuses.length < 3) rec.statuses.push(s.ok ? "online" : "offline")
      if (!byKey.has(key)) rec.sampledAt = run.sampledAt
      byKey.set(key, rec)
    }
  }
  const out = new Map<string, ServerSignal>()
  for (const [key, rec] of byKey) out.set(key, timelineToServerSignal(rec.name, rec.sampledAt, rec.statuses))
  return out
}

function mergeResourceAndRestServer(resource: ServerSignal, rest: ServerSignal): ServerSignal {
  const rUp = resource.status === "online"
  const tUp = rest.status === "online"
  if (tUp && !rUp) {
    const rBad = resource.status === "offline" || resource.status === "degraded"
    if (rBad) {
      return {
        name: rest.name,
        status: "online",
        prevStatus: "online",
        prev2Status: resource.status,
        sampledAt: rest.sampledAt,
      }
    }
    return rest
  }
  if (rUp && !tUp) return resource
  if (tUp && rUp) {
    if (rest.sampledAt >= resource.sampledAt) {
      if (resource.prevStatus === "offline" || resource.prevStatus === "degraded") {
        return {
          name: rest.name,
          status: "online",
          prevStatus: "online",
          prev2Status: resource.prevStatus === "degraded" ? "degraded" : "offline",
          sampledAt: rest.sampledAt,
        }
      }
      if (
        resource.prevStatus === "online" &&
        resource.prev2Status != null &&
        (resource.prev2Status === "offline" || resource.prev2Status === "degraded")
      ) {
        return {
          name: rest.name,
          status: "online",
          prevStatus: "online",
          prev2Status: resource.prev2Status,
          sampledAt: rest.sampledAt,
        }
      }
    }
    return rest.sampledAt >= resource.sampledAt ? rest : resource
  }
  return rest.sampledAt >= resource.sampledAt ? rest : resource
}

function collectProbeSignalsFromRuns(runs: PingRunSnapshot[]): ProbeSignal[] {
  const out = new Map<string, ProbeSignal>()
  for (const run of runs) {
    for (const p of run.probes ?? []) {
      const key = `${String(p.name ?? "").trim()} → ${String(p.target ?? "").trim()}`
      if (!key.trim() || out.has(key)) continue
      out.set(key, {
        key,
        status: p.status,
        rttMs: p.rttMs ?? null,
        lossPct: p.lossPct ?? null,
        sampledAt: run.sampledAt,
      })
    }
  }
  return [...out.values()]
}

function collectTrafficSignalsFromRuns(runs: TrafficRunSnapshot[]): TrafficServerSignal[] {
  const out = new Map<string, TrafficServerSignal>()
  for (const run of runs) {
    for (const s of run.servers ?? []) {
      const name = String(s.name ?? "").trim()
      if (!name) continue
      const key = normalizeNameKey(name)
      if (out.has(key)) continue
      out.set(key, {
        serverName: name,
        rxMbps: Number(s.sumRxMbps ?? 0),
        txMbps: Number(s.sumTxMbps ?? 0),
        sampledAt: run.sampledAt,
      })
    }
  }
  return [...out.values()]
}

function collectGreSignalsFromRuns(runs: GreBgpSnapshotRunSnapshot[]): GreTunnelSignal[] {
  const byTarget = new Map<string, GreTunnelSignal>()
  for (const run of runs) {
    const greRows = run.greTunnels ?? []
    for (const row of greRows) {
      const label = String(row.targetLabel ?? "").trim()
      if (!label) continue
      const prev = byTarget.get(label)
      if (!prev) {
        byTarget.set(label, {
          targetLabel: label,
          status: row.status,
          prevStatus: null,
        })
        continue
      }
      if (prev.prevStatus == null) prev.prevStatus = row.status
    }
  }
  return [...byTarget.values()]
}

function collectBgpSignalsFromRuns(runs: GreBgpSnapshotRunSnapshot[]): BgpPeerSignal[] {
  const byKey = new Map<string, BgpPeerSignal>()
  for (const run of runs) {
    const peers = run.bgpPeers ?? []
    for (const row of peers) {
      const key = String(row.key ?? "").trim()
      if (!key) continue
      const prev = byKey.get(key)
      if (!prev) {
        byKey.set(key, { key, state: String(row.state ?? "").trim(), prevState: null })
        continue
      }
      if (prev.prevState == null) prev.prevState = String(row.state ?? "").trim()
    }
  }
  return [...byKey.values()]
}

/** Собирает сигналы alert_engine напрямую из результатов collector jobs (`scheduler_runs.result_json`). */
export function buildSignalSnapshotFromCollectors(): SignalSnapshot {
  const sampledAt = new Date().toISOString()
  const resourceRuns = readLatestOkRunSnapshots("uptime_resources", 16) as ResourcesRunSnapshot[]
  const restRuns = readLatestOkRunSnapshots("servers_rest_ping", 16) as ServersRestPingRunSnapshot[]
  const pingRuns = readLatestOkRunSnapshots("uptime_ping", 24) as PingRunSnapshot[]
  const trafficRuns = readLatestOkRunSnapshots("traffic", 8) as TrafficRunSnapshot[]
  const greBgpRuns = readLatestOkRunSnapshots("gre_bgp", 8) as GreBgpSnapshotRunSnapshot[]

  const resourceByKey = collectServerSignalsFromResourceRuns(resourceRuns)
  const restByKey = collectServerSignalsFromRestRuns(restRuns)
  const serverKeys = new Set([...resourceByKey.keys(), ...restByKey.keys()])
  const servers: ServerSignal[] = []
  for (const key of serverKeys) {
    const resource = resourceByKey.get(key)
    const rest = restByKey.get(key)
    if (!rest) {
      if (resource) servers.push(resource)
      continue
    }
    if (!resource) {
      servers.push(rest)
      continue
    }
    servers.push(mergeResourceAndRestServer(resource, rest))
  }
  servers.sort((a, b) => a.name.localeCompare(b.name, undefined, { sensitivity: "base" }))

  return {
    sampledAt,
    servers,
    probes: collectProbeSignalsFromRuns(pingRuns),
    trafficByServer: collectTrafficSignalsFromRuns(trafficRuns),
    greTunnels: collectGreSignalsFromRuns(greBgpRuns),
    bgpSessions: collectBgpSignalsFromRuns(greBgpRuns),
  }
}