feat(traffic): показать аналитику IPFIX по серверам и интерфейсам

Резолвить ifIndex в имена RouterOS, дать вкладке Потоки ту же оболочку сервер/клиент/iface, что у обычного трафика, и обновлять срезы live без перезагрузки.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/src/routes/traffic-flow.ts

@@ -1,5 +1,6 @@
 import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
 import type { FastifyReply, FastifyRequest } from "fastify"
+import { env } from "../config.js"
 import {
   trafficFlowOverlayRequestSchema,
   trafficFlowSettingsPatchSchema,
@@ -12,12 +13,19 @@ import {
 } from "../services/traffic-flow-settings.js"
 import {
   getFlowListenerState,
-  listFlowTalkers,
   startTrafficFlowListener,
+  listFlowTalkers,
 } from "../services/traffic-flow-ingest.js"
+import {
+  buildFlowAnalytics,
+  listFlowClients,
+  listFlowExporters,
+} from "../services/traffic-flow-analytics.js"
 import { applyFlowOverlay } from "../services/traffic-flow-overlay.js"
 import { listTrafficFlowHostFiles } from "../services/traffic-flow-host-files.js"
 
+const LIVE_TICK_MS = 2000
+
 function rangeToMinutes(range: string | undefined): number {
   switch ((range ?? "5m").toLowerCase()) {
     case "5m": return 5
@@ -29,6 +37,22 @@ function rangeToMinutes(range: string | undefined): number {
   }
 }
 
+function parseId(raw: unknown): number | undefined {
+  if (raw == null || raw === "") return undefined
+  const n = Number.parseInt(String(raw), 10)
+  return Number.isFinite(n) ? n : undefined
+}
+
+function analyticsQuery(req: FastifyRequest) {
+  const q = req.query as { range?: string; serverId?: string; userId?: string; iface?: string }
+  return {
+    minutes: rangeToMinutes(q.range),
+    serverId: parseId(q.serverId),
+    userId: q.userId?.trim() || undefined,
+    iface: q.iface?.trim() || undefined,
+  }
+}
+
 async function sendFlowTalkers(req: FastifyRequest, reply: FastifyReply) {
   const q = req.query as { range?: string }
   return reply.send(listFlowTalkers(rangeToMinutes(q.range)))
@@ -58,6 +82,28 @@ async function applyOverlayHandler(req: FastifyRequest, reply: FastifyReply) {
   }
 }
 
+function writeSse(raw: NodeJS.WritableStream, event: string, data: unknown) {
+  raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
+}
+
+function sleep(ms: number, signal: AbortSignal): Promise<void> {
+  return new Promise((resolve, reject) => {
+    if (signal.aborted) {
+      reject(new Error("aborted"))
+      return
+    }
+    const timer = setTimeout(() => {
+      signal.removeEventListener("abort", onAbort)
+      resolve()
+    }, ms)
+    const onAbort = () => {
+      clearTimeout(timer)
+      reject(new Error("aborted"))
+    }
+    signal.addEventListener("abort", onAbort, { once: true })
+  })
+}
+
 const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => {
   app.get("/traffic/flow/settings", async (_req, reply) => {
     return reply.send(toTrafficFlowSettingsDto(getFlowListenerState()))
@@ -94,6 +140,63 @@ const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => {
 
   app.get("/traffic/flow", sendFlowTalkers)
   app.get("/traffic/flows", sendFlowTalkers)
+
+  app.get("/traffic/flow/exporters", async (req, reply) => {
+    const q = req.query as { range?: string }
+    return reply.send(listFlowExporters(rangeToMinutes(q.range)))
+  })
+
+  app.get("/traffic/flow/clients", async (req, reply) => {
+    const q = req.query as { range?: string }
+    return reply.send(listFlowClients(rangeToMinutes(q.range)))
+  })
+
+  app.get("/traffic/flow/analytics", async (req, reply) => {
+    return reply.send(buildFlowAnalytics(analyticsQuery(req)))
+  })
+
+  app.get("/traffic/flow/live", async (req, reply) => {
+    const query = analyticsQuery(req)
+    const abort = new AbortController()
+    const onClose = () => abort.abort()
+    req.raw.on("close", onClose)
+
+    reply.hijack()
+    req.raw.setTimeout(0)
+    reply.raw.setTimeout(0)
+    const origin = typeof req.headers.origin === "string" ? req.headers.origin : ""
+    const allowed = env.CORS_ORIGIN
+    const sseHeaders: Record<string, string> = {
+      "Content-Type": "text/event-stream; charset=utf-8",
+      "Cache-Control": "no-cache, no-transform",
+      Connection: "keep-alive",
+      "X-Accel-Buffering": "no",
+    }
+    if (origin && (allowed === "*" || allowed === origin)) {
+      sseHeaders["Access-Control-Allow-Origin"] = origin
+      sseHeaders["Access-Control-Allow-Credentials"] = "true"
+      sseHeaders["Access-Control-Allow-Headers"] = "Authorization, Accept"
+      sseHeaders.Vary = "Origin"
+    }
+    reply.raw.writeHead(200, sseHeaders)
+    reply.raw.write(":\n\n")
+
+    try {
+      while (!abort.signal.aborted) {
+        writeSse(reply.raw, "sample", buildFlowAnalytics(query))
+        await sleep(LIVE_TICK_MS, abort.signal)
+      }
+    } catch {
+      /* abort / disconnect */
+    } finally {
+      req.raw.off("close", onClose)
+      try {
+        reply.raw.end()
+      } catch {
+        /* already closed */
+      }
+    }
+  })
 }
 
 export default trafficFlowRoutes