MikrotikManager/backend/src/services/traffic-rate.ts

/** Чистые формулы трафика: дельты счётчиков, корзины series, monitor-traffic. */

export const SERIES_POINTS = 60

export interface TrafficSampleLike {
  interfaceName: string
  peerPublicKey?: string
  sampledAt: string
  rxBytes: number
  txBytes: number
  rxBps: number
  txBps: number
  running: boolean
  disabled: boolean
}

export interface RatePoint {
  t: number
  rxMbps: number
  txMbps: number
}

export interface TrafficObservations {rxNow:boolean;txNow:boolean;rxSeries:boolean[];txSeries:boolean[];sampledAt:string|null}

export interface BuiltTrafficSeries {
  observations?:TrafficObservations
  rxNow: number
  txNow: number
  rxPeak: number
  txPeak: number
  rxTotalGiB: number
  txTotalGiB: number
  sessions: number
  rxSeries: number[]
  txSeries: number[]
}

export interface MonitorLiveSample {
  rxMbps: number
  txMbps: number
  at: string
}

export function isLoopbackName(name: string): boolean {
  return /^(lo|loopback)(\d+)?$/i.test(name.trim())
}

export function shouldIncludeIface(
  name: string,
  running: boolean,
  disabled: boolean,
  onlyInterface?: string,
): boolean {
  if (onlyInterface) return name === onlyInterface
  if (isLoopbackName(name)) return false
  return running && !disabled
}

export function bpsToMbps(bps: number): number {
  if (!Number.isFinite(bps) || bps <= 0) return 0
  return Math.round((bps / 1_000_000) * 1000) / 1000
}

/** bits/s из соседних счётчиков. null = нельзя (Δt≤0 или сброс). */
export function rateBpsFromDelta(
  prevBytes: number,
  nextBytes: number,
  prevAtMs: number,
  nextAtMs: number,
): number | null {
  if(![prevBytes,nextBytes,prevAtMs,nextAtMs].every(Number.isFinite)||prevBytes<0||nextBytes<0)return null
  const dtSec = (nextAtMs - prevAtMs) / 1000
  if (!(dtSec > 0) || !Number.isFinite(dtSec)) return null
  if (nextBytes < prevBytes) return null
  const rate=Math.round(((nextBytes - prevBytes) * 8) / dtSec)
  return Number.isFinite(rate)?rate:null
}

export function bucketAvg(
  points: Array<{ t: number; v: number }>,
  rangeStartMs: number,
  rangeEndMs: number,
  target = SERIES_POINTS,
): number[] {
  const buckets = Array.from({ length: target }, () => 0)
  const counts = Array.from({ length: target }, () => 0)
  const span = rangeEndMs - rangeStartMs
  if (span <= 0 || points.length === 0) return buckets
  for (const p of points) {
    const ratio = (p.t - rangeStartMs) / span
    const i = Math.min(target - 1, Math.max(0, Math.floor(ratio * target)))
    buckets[i] += p.v
    counts[i] += 1
  }
  return buckets.map((sum, i) => (counts[i] > 0 ? sum / counts[i] : 0))
}

function parseIsoMs(iso: string): number {
  return Date.parse(iso)
}

export function sampleSeriesKey(interfaceName: string, peerPublicKey = ""): string {
  return `${interfaceName}\0${peerPublicKey}`
}

export function buildTrafficFromSamples(
  rows: TrafficSampleLike[],
  rangeStartMs: number,
  rangeEndMs: number,
  onlyInterface?: string | readonly string[],
  peerPublicKey?: string,
  includeObservations=false,
): BuiltTrafficSeries {
  const empty: BuiltTrafficSeries = {
    rxNow: 0,
    txNow: 0,
    rxPeak: 0,
    txPeak: 0,
    rxTotalGiB: 0,
    txTotalGiB: 0,
    sessions: 0,
    rxSeries: Array.from({ length: SERIES_POINTS }, () => 0),
    txSeries: Array.from({ length: SERIES_POINTS }, () => 0),
  }
  const noObservations:TrafficObservations={rxNow:false,txNow:false,rxSeries:Array(SERIES_POINTS).fill(false),txSeries:Array(SERIES_POINTS).fill(false),sampledAt:null}
  if (rows.length === 0) return includeObservations?{...empty,observations:noObservations}:empty

  const bySeries = new Map<string, TrafficSampleLike[]>()
  for (const r of rows) {
    const peer = r.peerPublicKey ?? ""
    const arr = bySeries.get(sampleSeriesKey(r.interfaceName, peer)) ?? []
    arr.push(r)
    bySeries.set(sampleSeriesKey(r.interfaceName, peer), arr)
  }

  const allowList = Array.isArray(onlyInterface)
    ? onlyInterface
    : (typeof onlyInterface === "string" ? [onlyInterface] : null)

  const rxPoints: Array<{ t: number; v: number }> = []
  const txPoints: Array<{ t: number; v: number }> = []
  const byTs = new Map<number, { rx: number; tx: number;rxObserved:boolean;txObserved:boolean;count:number }>()
  let includedSeries=0,lastAttemptAt:number|null=null

  let rxBytesDelta = 0
  let txBytesDelta = 0
  let sessions = 0

  for (const [key, arr] of bySeries) {
    const sep = key.indexOf("\0")
    const name = sep >= 0 ? key.slice(0, sep) : key
    const peer = sep >= 0 ? key.slice(sep + 1) : ""
    if (allowList) {
      if (!allowList.includes(name)) continue
    } else if (isLoopbackName(name)) {
      continue
    }
    if (peerPublicKey === undefined) {
      if (peer !== "") continue
    } else if (peer !== peerPublicKey) {
      continue
    }

    const sorted = [...arr].sort((a, b) => a.sampledAt.localeCompare(b.sampledAt))
    const last = sorted[sorted.length - 1]
    if (!last) continue
    if (!allowList && (!last.running || last.disabled)) continue

    includedSeries++
    if (last.running && !last.disabled) sessions += 1

    const first = sorted[0]
    if (first) {
      const dRx = last.rxBytes - first.rxBytes
      const dTx = last.txBytes - first.txBytes
      rxBytesDelta += dRx >= 0 ? dRx : last.rxBytes
      txBytesDelta += dTx >= 0 ? dTx : last.txBytes
    }

    for (let i = 1; i < sorted.length; i++) {
      const prev = sorted[i - 1]
      const cur = sorted[i]
      if (!prev || !cur) continue
      if (!allowList && (!cur.running || cur.disabled)) continue
      const t0 = parseIsoMs(prev.sampledAt)
      const t1 = parseIsoMs(cur.sampledAt)
      if(Number.isFinite(Date.parse(cur.sampledAt))&&t1>=rangeStartMs&&t1<=rangeEndMs)lastAttemptAt=Math.max(lastAttemptAt??t1,t1)
      const rxBps = rateBpsFromDelta(prev.rxBytes, cur.rxBytes, t0, t1)
      const txBps = rateBpsFromDelta(prev.txBytes, cur.txBytes, t0, t1)
      if (rxBps == null && txBps == null) continue
      const rxMbps = bpsToMbps(rxBps ?? 0)
      const txMbps = bpsToMbps(txBps ?? 0)
      const acc = byTs.get(t1) ?? { rx: 0, tx: 0,rxObserved:true,txObserved:true,count:0 }
      acc.count++;acc.rxObserved&&=rxBps!==null;acc.txObserved&&=txBps!==null
      acc.rx += rxMbps
      acc.tx += txMbps
      byTs.set(t1, acc)
    }
  }

  for (const [t, v] of byTs) {
    rxPoints.push({ t, v: v.rx })
    txPoints.push({ t, v: v.tx })
  }

  const rxSeries = bucketAvg(rxPoints, rangeStartMs, rangeEndMs)
  const txSeries = bucketAvg(txPoints, rangeStartMs, rangeEndMs)
  const lastTs = [...byTs.keys()].sort((a, b) => a - b).at(-1)
  const last = lastTs != null ? byTs.get(lastTs) : undefined
  const rxPeak = rxPoints.reduce((m, p) => Math.max(m, p.v), 0)
  const txPeak = txPoints.reduce((m, p) => Math.max(m, p.v), 0)

  return {
    rxNow: last?.rx ?? 0,
    txNow: last?.tx ?? 0,
    rxPeak,
    txPeak,
    rxTotalGiB: Number((rxBytesDelta / (1024 ** 3)).toFixed(1)),
    txTotalGiB: Number((txBytesDelta / (1024 ** 3)).toFixed(1)),
    sessions,
    rxSeries,
    txSeries,
    ...(includeObservations?{observations:{
      rxNow:lastAttemptAt!==null&&lastTs===lastAttemptAt&&!!last?.rxObserved&&last.count===includedSeries,
      txNow:lastAttemptAt!==null&&lastTs===lastAttemptAt&&!!last?.txObserved&&last.count===includedSeries,
      rxSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.rxObserved&&v.count===includedSeries&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1),
      txSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.txObserved&&v.count===includedSeries&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1),
      sampledAt:lastAttemptAt===null?null:new Date(lastAttemptAt).toISOString(),
    }}:{}),
  }
}

export function mergeBuiltTraffic(parts: BuiltTrafficSeries[]): BuiltTrafficSeries {
  const empty: BuiltTrafficSeries = {
    rxNow: 0,
    txNow: 0,
    rxPeak: 0,
    txPeak: 0,
    rxTotalGiB: 0,
    txTotalGiB: 0,
    sessions: 0,
    rxSeries: Array.from({ length: SERIES_POINTS }, () => 0),
    txSeries: Array.from({ length: SERIES_POINTS }, () => 0),
  }
  if (parts.length === 0) return empty
  const acc = {
    rxNow: 0,
    txNow: 0,
    rxPeak: 0,
    txPeak: 0,
    rxTotalGiB: 0,
    txTotalGiB: 0,
    sessions: 0,
    rxSeries: Array.from({ length: SERIES_POINTS }, () => 0),
    txSeries: Array.from({ length: SERIES_POINTS }, () => 0),
  }
  for (const p of parts) {
    acc.rxNow += p.rxNow
    acc.txNow += p.txNow
    acc.rxPeak += p.rxPeak
    acc.txPeak += p.txPeak
    acc.rxTotalGiB += p.rxTotalGiB
    acc.txTotalGiB += p.txTotalGiB
    acc.sessions += p.sessions
    for (let i = 0; i < SERIES_POINTS; i++) {
      acc.rxSeries[i] += p.rxSeries[i] ?? 0
      acc.txSeries[i] += p.txSeries[i] ?? 0
    }
  }
  acc.rxTotalGiB = Number(acc.rxTotalGiB.toFixed(1))
  acc.txTotalGiB = Number(acc.txTotalGiB.toFixed(1))
  return acc
}

export function parseMonitorTraffic(
  raw: unknown,
  opts?: { onlyInterface?: string },
): MonitorLiveSample {
  const items = Array.isArray(raw) ? raw : raw != null ? [raw] : []
  let rxBps = 0
  let txBps = 0
  for (const item of items) {
    if (!item || typeof item !== "object") continue
    const rec = item as Record<string, unknown>
    const name = String(rec.name ?? rec.interface ?? "")
    if (opts?.onlyInterface) {
      if (name && name !== opts.onlyInterface) continue
    } else if (isLoopbackName(name)) {
      continue
    }
    rxBps += Number.parseFloat(String(rec["rx-bits-per-second"] ?? 0)) || 0
    txBps += Number.parseFloat(String(rec["tx-bits-per-second"] ?? 0)) || 0
  }
  return {
    rxMbps: bpsToMbps(rxBps),
    txMbps: bpsToMbps(txBps),
    at: new Date().toISOString(),
  }
}