mihomo-patchbay/web/telemetry.ts

/**
 * 常驻的遥测采集：流量曲线、内核内存、每路入站的吞吐。
 *
 * 住在模块级单例里而不是仪表盘组件里，是因为历史要跨页面留存——采集跟着组件
 * 走的话，切走一次曲线就清零一次。应用启动后由外壳调一次 startTelemetry()，
 * 之后跟着 kernel.online 自己起落：上线接上，掉线断开但**不清历史**，断档在
 * 时间轴上自然呈现为一段空白。
 *
 * 数据来源两路：
 *   - /traffic、/memory 流，每秒一行（内核源码 hub/route/server.go:386-414）
 *   - /connections 快照 3 秒一轮询，按连接逐条求差聚合出每路的吞吐
 *     （快照本身没有流式 HTTP 版本，见 hub/route/connections.go:24-28）
 */
import { ref, watch } from 'vue'
import {
  ROUTE_POLL_MS,
  ROUTE_WINDOW_MS,
  TRAFFIC_WINDOW_MS,
  pushWindow,
  routeDeltas,
} from '@shared/telemetry.ts'
import type { ConnTotals, RouteSample, TrafficPoint } from '@shared/telemetry.ts'
import type { ExitProbe } from '@shared/host-info.ts'
import { api } from './api'
import { kernel } from './runtime'
import { readLines } from './stream'

/** 流量曲线，最近 30 分钟。up/down 是内核报的每秒字节数。 */
export const traffic = ref<TrafficPoint[]>([])

/** 内核报的累计上下行字节数。没收到过是 null。 */
export const totals = ref<{ up: number; down: number } | null>(null)

/** 内核自己的内存占用。 */
export const kernelMemory = ref<number | null>(null)

/** 每路入站的吞吐样本，键是内核连接快照里的 inboundName。 */
export const routeSamples = ref<Record<string, RouteSample[]>>({})

/** 当前连接数。 */
export const connCount = ref<number | null>(null)

/** 流当前接着没有。断流期间仪表盘的「当前值」要显示成没数据，而不是冻在最后一秒。 */
export const live = ref(false)

/** 每个端口的公网 IP 检测结果，会话内缓存。键是端口号。 */
export const exitProbes = ref<Record<number, { state: 'loading' | 'done'; result?: ExitProbe }>>({})

/** 采集的时钟，每次轮询走一格。活跃判定这类「相对现在」的计算依赖它保持新鲜。 */
export const tick = ref(Date.now())

const num = (value: unknown): number =>
  typeof value === 'number' && Number.isFinite(value) ? value : 0

let abort: AbortController | null = null
let pollTimer = 0
let retryTimer = 0
let prevTotals: ConnTotals = new Map()
let prevAt = 0

async function consume(
  path: string,
  ctrl: AbortController,
  onLine: (payload: Record<string, unknown>) => void,
) {
  const body = await api.openStream(path, ctrl.signal)
  for await (const line of readLines(body)) {
    // 偶发的坏行跳过就行，抛出去会杀掉整个流、进 5 秒重试循环
    let payload: Record<string, unknown>
    try {
      payload = JSON.parse(line) as Record<string, unknown>
    } catch {
      continue
    }
    onLine(payload)
  }
}

function startStreams() {
  const ctrl = new AbortController()
  abort = ctrl
  live.value = true
  const run = (path: string, onLine: (payload: Record<string, unknown>) => void) =>
    void consume(path, ctrl, onLine)
      .catch(() => {})
      // 这两个流本身永不结束，正常返回也意味着断了
      .finally(() => {
        if (ctrl.signal.aborted) return
        // 一条断了把另一条也收掉，然后走统一的重连节奏
        ctrl.abort()
        live.value = false
        scheduleRetry()
      })

  run('/traffic', (p) => {
    traffic.value = pushWindow(
      traffic.value,
      { t: Date.now(), up: num(p.up), down: num(p.down) },
      TRAFFIC_WINDOW_MS,
    )
    totals.value = { up: num(p.upTotal), down: num(p.downTotal) }
  })
  run('/memory', (p) => {
    kernelMemory.value = num(p.inuse)
  })
}

/**
 * 断流不等于内核没了：保存配置触发的重载会掐断流，内核几秒内就回来、online 从未
 * 翻转。5 秒后重试，失败会再次走到这里形成节奏；真掉线交给下面 online 的 watch。
 */
function scheduleRetry() {
  window.clearTimeout(retryTimer)
  retryTimer = window.setTimeout(() => {
    if (kernel.value.online && !live.value) startStreams()
  }, 5_000)
}

async function pollConnections() {
  const now = Date.now()
  try {
    const snapshot = await api.connections()
    const conns = snapshot.connections ?? []
    connCount.value = conns.length
    const { deltas, next } = routeDeltas(prevTotals, conns)
    const ms = prevAt ? now - prevAt : ROUTE_POLL_MS
    prevTotals = next
    prevAt = now

    const merged = { ...routeSamples.value }
    for (const [key, delta] of Object.entries(deltas)) {
      merged[key] = pushWindow(
        merged[key] ?? [],
        { t: now, ms, up: delta.up, down: delta.down },
        ROUTE_WINDOW_MS,
      )
    }
    // 有历史但这一轮没有连接的路，补零样本让走势落回去而不是冻住；
    // 整个窗口都归零之后把这路忘掉，改名、删除的入站才不会在内存里越攒越多
    for (const key of Object.keys(merged)) {
      if (key in deltas) continue
      const samples = pushWindow(merged[key]!, { t: now, ms, up: 0, down: 0 }, ROUTE_WINDOW_MS)
      if (samples.every((s) => s.up === 0 && s.down === 0)) delete merged[key]
      else merged[key] = samples
    }
    routeSamples.value = merged
  } catch {
    // 单轮失败就跳过。持续失败说明内核掉了，online 的 watch 会把轮询停掉
  } finally {
    tick.value = now
  }
}

function attach() {
  if (!abort) startStreams()
  if (!pollTimer) {
    void pollConnections()
    pollTimer = window.setInterval(() => void pollConnections(), ROUTE_POLL_MS)
  }
}

function detach() {
  abort?.abort()
  abort = null
  live.value = false
  window.clearInterval(pollTimer)
  window.clearTimeout(retryTimer)
  pollTimer = 0
  prevTotals = new Map()
  prevAt = 0
}

let started = false

/** 外壳登录成功后调一次。之后采集跟着内核在线状态自己起落。 */
export function startTelemetry(): void {
  if (started) return
  started = true
  watch(
    () => kernel.value.online,
    (online) => {
      if (online) attach()
      else detach()
    },
    { immediate: true },
  )
}

/** 检测某个端口的公网 IP。结果缓存到会话结束，force 重测。 */
export async function probeExit(port: number, force = false): Promise<void> {
  const cached = exitProbes.value[port]
  if (!force && cached && (cached.state === 'loading' || cached.result?.ok)) return
  exitProbes.value = { ...exitProbes.value, [port]: { state: 'loading' } }
  const result = await api
    .probeExit(port)
    .catch((error: unknown) => ({ ok: false, message: String(error) }) as ExitProbe)
  exitProbes.value = { ...exitProbes.value, [port]: { state: 'done', result } }
}