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

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

Co-authored-by: Cursor <[email protected]>

app/(main)/traffic/page.tsx

@@ -17,13 +17,13 @@ import { cn } from "@/lib/utils"
 import { fmtGB, fmtRate } from "@/lib/fmt-rate"
 import { useDataSource } from "@/lib/data-source"
 import { useTrafficLive } from "@/hooks/use-traffic-live"
+import { useFlowLive } from "@/hooks/use-flow-live"
 import { requestJson } from "@/shared/api/http-client"
-import { getTrafficFlows } from "@/shared/api/traffic-flow"
+import { getFlowAnalytics, getFlowClients, getFlowExporters, getTrafficFlows } from "@/shared/api/traffic-flow"
 import { listServers } from "@/shared/api/servers"
-import { TrafficFlowsDataGrid } from "@/components/data-grids/traffic-flows-data-grid"
 import { FlowOverlaySheet } from "@/components/traffic/flow-overlay-sheet"
-import { DataPageCard } from "@/components/data-page-card"
-import type { FlowStatsDto } from "@mmapp/contracts/traffic-flow"
+import { FlowAnalyticsDetail, FlowEntityCardView } from "@/components/traffic/flow-analytics-panel"
+import type { FlowAnalyticsDto, FlowEntityCard, FlowStatsDto } from "@mmapp/contracts/traffic-flow"
 import type { ServerRead } from "@mmapp/contracts/servers"
 import { Badge } from "@/components/reui/badge"
 import {
@@ -50,6 +50,31 @@ function addSeries(a: number[], b: number[]): number[] {
   return a.map((v, i) => v + (b[i] ?? 0))
 }
 
+function flowIngestLine(stats: FlowStatsDto | null): string | null {
+  if (!stats) return null
+  const listener = stats.listenerBound
+    ? (stats.listenerAddress ?? "слушает")
+    : "не слушает"
+  const last = stats.lastDatagramAt
+    ? new Date(stats.lastDatagramAt).toLocaleString("ru-RU")
+    : "—"
+  const exporter = stats.lastExporterIp ? ` · ${stats.lastExporterIp}` : ""
+  const err = stats.lastError ? ` · ${stats.lastError}` : ""
+  return `Коллектор: ${listener} · пакеты ${stats.packetsReceived ?? 0} · последний ${last}${exporter}${err}`
+}
+
+function flowEmptyHint(stats: FlowStatsDto | null): string | undefined {
+  if (!stats) return undefined
+  if (stats.lastError) return stats.lastError
+  if (stats.packetsReceived) {
+    return `IPFIX приходит (${stats.lastExporterIp ?? "экспортёр"}), но разговоры ещё не записаны.`
+  }
+  if (stats.listenerBound === false) {
+    return "Коллектор UDP не слушает. Подключите JH ещё раз — ingest включится автоматически."
+  }
+  return "IPFIX ещё не доходит до коллектора. На jump-host у target Src должен быть 0.0.0.0 (авто). На хосте MM проверьте bind 10.255.254.1:4739 после wg-flow."
+}
+
 // ─── data model ───────────────────────────────────────────────────────────────
 
 interface BoundIfaceTraffic {
@@ -284,6 +309,7 @@ const TRAFFIC_RANGE_LABELS: Record<Range, string> = {
 }
 
 type GroupMode = "servers" | "users" | "ifaces" | "flows"
+type FlowScope = "servers" | "users"
 type SortField = "rx" | "tx" | "name" | "sessions"
 type SortDir   = "asc" | "desc"
 
@@ -715,7 +741,7 @@ const SORT_FIELDS: Array<{ field: SortField; label: string; modesOnly?: GroupMod
   { field: "rx",       label: "RX" },
   { field: "tx",       label: "TX" },
   { field: "name",     label: "Имя" },
-  { field: "sessions", label: "Сессий", modesOnly: ["servers", "users"] },
+  { field: "sessions", label: "Сессий", modesOnly: ["servers", "users", "flows"] },
 ]
 
 export default function TrafficPage() {
@@ -740,6 +766,11 @@ export default function TrafficPage() {
   const [liveUsers, setLiveUsers] = useState<UserTraffic[]>([])
   const [liveBoundIfaces, setLiveBoundIfaces] = useState<BoundIfaceTraffic[]>([])
   const [flowStats, setFlowStats] = useState<FlowStatsDto | null>(null)
+  const [flowScope, setFlowScope] = useState<FlowScope>("servers")
+  const [flowExporters, setFlowExporters] = useState<FlowEntityCard[]>([])
+  const [flowClients, setFlowClients] = useState<FlowEntityCard[]>([])
+  const [flowAnalytics, setFlowAnalytics] = useState<FlowAnalyticsDto | null>(null)
+  const [flowIface, setFlowIface] = useState("__all__")
   const [overlayOpen, setOverlayOpen] = useState(false)
   const [catalogServers, setCatalogServers] = useState<ServerRead[]>([])
   const effectiveMode: GroupMode = groupMode
@@ -749,6 +780,15 @@ export default function TrafficPage() {
     serverId: selectedId,
     iface: selectedIface,
   })
+  const flowLiveEnabled = isLive && effectiveMode === "flows" && Boolean(selectedId)
+  const { sample: flowLiveSample, error: flowLiveError } = useFlowLive({
+    enabled: flowLiveEnabled,
+    backendUrl,
+    range,
+    serverId: flowScope === "servers" ? selectedId : undefined,
+    userId: flowScope === "users" ? selectedId : undefined,
+    iface: flowIface,
+  })
 
   const toLiveServer = (s: LiveTrafficServer): ServerTraffic => {
     return {
@@ -837,20 +877,50 @@ export default function TrafficPage() {
     setLiveBusy(true)
     setLiveError(null)
     try {
-      const stats = await getTrafficFlows(backendUrl, range)
+      const [stats, exporters, clients] = await Promise.all([
+        getTrafficFlows(backendUrl, range),
+        getFlowExporters(backendUrl, range),
+        getFlowClients(backendUrl, range),
+      ])
       setFlowStats(stats)
+      setFlowExporters(exporters.exporters)
+      setFlowClients(clients.clients)
+      setSelectedId((prev) => {
+        const list = flowScope === "users" ? clients.clients : exporters.exporters
+        if (list.some((x) => x.id === prev)) return prev
+        return list[0]?.id ?? ""
+      })
     } catch (e) {
       setLiveError(e instanceof Error ? e.message : "Не удалось загрузить потоки")
     } finally {
       setLiveBusy(false)
     }
-  }, [isLive, backendUrl, range])
+  }, [isLive, backendUrl, range, flowScope])
 
   useEffect(() => {
     if (!isLive || effectiveMode !== "flows") return
     void loadFlows()
+    const t = window.setInterval(() => { void loadFlows() }, 5000)
+    return () => window.clearInterval(t)
   }, [isLive, effectiveMode, loadFlows])
 
+  useEffect(() => {
+    if (!isLive || effectiveMode !== "flows" || !selectedId) {
+      setFlowAnalytics(null)
+      return
+    }
+    void getFlowAnalytics(backendUrl, {
+      range,
+      serverId: flowScope === "servers" ? selectedId : undefined,
+      userId: flowScope === "users" ? selectedId : undefined,
+      iface: flowIface,
+    }).then(setFlowAnalytics).catch(() => setFlowAnalytics(null))
+  }, [isLive, effectiveMode, selectedId, range, flowScope, flowIface, backendUrl])
+
+  useEffect(() => {
+    setFlowIface("__all__")
+  }, [selectedId, flowScope])
+
   useEffect(() => {
     if (!isLive) return
     void listServers(backendUrl).then(setCatalogServers).catch(() => setCatalogServers([]))
@@ -899,6 +969,11 @@ export default function TrafficPage() {
     if (next === "servers") setSelectedId(activeServerTraffic[0]?.id ?? "srv1")
     else if (next === "users") setSelectedId((isLive ? liveUsers : userTraffic)[0]?.id ?? "u1")
     else if (next === "ifaces") setSelectedId((isLive ? liveBoundIfaces : boundIfaces)[0]?.id ?? "")
+    else if (next === "flows") {
+      setFlowScope("servers")
+      setFlowIface("__all__")
+      setSelectedId(flowExporters[0]?.id ?? "")
+    }
     setSortField("rx")
     setSortDir("desc")
     setSearch("")
@@ -953,6 +1028,22 @@ export default function TrafficPage() {
       })
   }, [sortField, sortDir, q, activeBoundIfaces])
 
+  const flowCards = flowScope === "users" ? flowClients : flowExporters
+  const sortedFlowCards = useMemo(() => {
+    return [...flowCards]
+      .filter((c) => !q || c.name.toLowerCase().includes(q) || c.subtitle.toLowerCase().includes(q) || c.site.toLowerCase().includes(q))
+      .sort((a, b) => {
+        let v = 0
+        if (sortField === "rx") v = a.rxNow - b.rxNow
+        else if (sortField === "tx") v = a.txNow - b.txNow
+        else if (sortField === "name") v = a.name.localeCompare(b.name)
+        else if (sortField === "sessions") v = a.sessions - b.sessions
+        return sortDir === "desc" ? -v : v
+      })
+  }, [flowCards, q, sortField, sortDir])
+  const selFlowCard = sortedFlowCards.find((c) => c.id === selectedId) ?? sortedFlowCards[0] ?? null
+  const displayedFlow = flowLiveSample ?? flowAnalytics
+
   const selServer = useMemo(() => activeServerTraffic.find(s => s.id === selectedId) ?? activeServerTraffic[0], [selectedId, activeServerTraffic])
   const detailServer = liveDetailServer?.id === selectedId ? liveDetailServer : selServer
   const selUser   = useMemo(() => activeUserTraffic.find(u => u.id === selectedId)   ?? activeUserTraffic[0],   [selectedId, activeUserTraffic])
@@ -969,33 +1060,36 @@ export default function TrafficPage() {
   const peakTx  = kpiSource.reduce((a, s) => Math.max(a, s.txPeak), 0)
 
   const visibleSortFields = SORT_FIELDS.filter(s => !s.modesOnly || s.modesOnly.includes(effectiveMode))
+  const ingestLine = flowIngestLine(flowStats)
+  const flowError = liveError || flowLiveError
 
   const flowKpiItems = [
     {
       id: "exporters",
       label: "Экспортёры",
-      value: String(flowStats?.exportersOnline ?? 0),
+      value: String(flowExporters.length),
       icon: <ServerIcon className="size-4" />,
       iconClassName: "text-info",
     },
     {
-      id: "bytes",
-      label: "Байт/мин",
-      value: flowStats ? fmtRate((flowStats.bytesPerMin * 8) / 1_000_000) : "—",
+      id: "bps",
+      label: "Скорость",
+      value: displayedFlow ? fmtRate(displayedFlow.bpsNow / 1_000_000) : (flowStats ? fmtRate((flowStats.bytesPerMin * 8) / 1_000_000) : "—"),
+      hint: displayedFlow?.live ? "live" : undefined,
       icon: <ActivityIcon className="size-4" />,
       iconClassName: "text-success",
     },
     {
       id: "src",
       label: "Уник. src",
-      value: String(flowStats?.uniqueSrc ?? 0),
+      value: String(displayedFlow?.uniqueSrc ?? flowStats?.uniqueSrc ?? 0),
       icon: <ArrowUpIcon className="size-4" />,
       iconClassName: "text-muted-foreground",
     },
     {
       id: "proto",
       label: "Топ протокол",
-      value: flowStats?.topProto ?? "—",
+      value: displayedFlow?.topProto ?? flowStats?.topProto ?? "—",
       icon: <GitBranchIcon className="size-4" />,
       iconClassName: "text-warning",
     },
@@ -1081,41 +1175,106 @@ export default function TrafficPage() {
           {effectiveMode === "flows" ? (
             <div className="flex flex-col gap-3">
               <div className="flex items-center justify-between gap-2 flex-wrap">
-                <p className="text-sm text-muted-foreground">
-                  IPFIX top-разговоры. Счётчики интерфейсов — в режимах Серверы / Клиенты / Интерфейсы.
+                <p className="text-xs text-muted-foreground font-mono truncate min-w-0">
+                  {ingestLine ?? "IPFIX коллектор"}
                 </p>
-                <div className="flex items-center gap-2 flex-wrap">
+                <Button size="sm" onClick={() => setOverlayOpen(true)} disabled={!isLive}>
+                  <PlusIcon className="size-4" />
+                  Подключить JH
+                </Button>
+              </div>
+              {flowError && (
+                <div className="text-xs text-destructive bg-destructive/10 border border-destructive/20 rounded-md px-3 py-2">
+                  {flowError}
+                </div>
+              )}
+              <div className="grid grid-cols-[300px_1fr] gap-5 items-start">
+                <div className="flex flex-col gap-3">
                   <div className="flex gap-1">
-                    {TRAFFIC_RANGE_KEYS.map((key) => (
+                    <button
+                      type="button"
+                      onClick={() => { setFlowScope("servers"); setFlowIface("__all__") }}
+                      className={cn(
+                        "text-[10px] px-2 py-0.5 rounded border transition-colors",
+                        flowScope === "servers"
+                          ? "border-primary bg-primary/10 text-primary font-medium"
+                          : "border-border text-muted-foreground hover:text-foreground",
+                      )}
+                    >
+                      Серверы
+                    </button>
+                    <button
+                      type="button"
+                      onClick={() => { setFlowScope("users"); setFlowIface("__all__") }}
+                      className={cn(
+                        "text-[10px] px-2 py-0.5 rounded border transition-colors",
+                        flowScope === "users"
+                          ? "border-primary bg-primary/10 text-primary font-medium"
+                          : "border-border text-muted-foreground hover:text-foreground",
+                      )}
+                    >
+                      Клиенты
+                    </button>
+                  </div>
+                  <div className="relative">
+                    <SearchIcon className="absolute left-2.5 top-1/2 -translate-y-1/2 size-3.5 text-muted-foreground pointer-events-none" />
+                    <input
+                      type="text"
+                      value={search}
+                      onChange={(e) => setSearch(e.target.value)}
+                      placeholder="Поиск…"
+                      className="w-full pl-8 pr-3 h-8 text-xs bg-muted/50 border border-border rounded-md outline-none focus:border-primary transition-colors"
+                    />
+                  </div>
+                  <div className="flex gap-1 flex-wrap">
+                    <span className="text-[10px] text-muted-foreground self-center mr-0.5">Сортировка:</span>
+                    {visibleSortFields.map(({ field, label }) => (
                       <button
-                        key={key}
+                        key={field}
                         type="button"
-                        onClick={() => setRange(key)}
+                        onClick={() => toggleSort(field)}
                         className={cn(
                           "text-[10px] px-2 py-0.5 rounded border transition-colors",
-                          range === key
+                          sortField === field
                             ? "border-primary bg-primary/10 text-primary font-medium"
                             : "border-border text-muted-foreground hover:text-foreground",
                         )}
                       >
-                        {TRAFFIC_RANGE_LABELS[key]}
+                        {label}{sortField === field ? (sortDir === "desc" ? " ↓" : " ↑") : ""}
                       </button>
                     ))}
                   </div>
-                  <Button size="sm" onClick={() => setOverlayOpen(true)} disabled={!isLive}>
-                    <PlusIcon className="size-4" />
-                    Подключить JH
-                  </Button>
+                  <div className="flex flex-col gap-2">
+                    {sortedFlowCards.map((card) => (
+                      <FlowEntityCardView
+                        key={card.id}
+                        card={card}
+                        selected={card.id === (selFlowCard?.id ?? selectedId)}
+                        onClick={() => setSelectedId(card.id)}
+                      />
+                    ))}
+                    {sortedFlowCards.length === 0 ? (
+                      <p className="text-xs text-muted-foreground">
+                        {flowEmptyHint(flowStats) ?? "Нет экспортёров IPFIX. Подключите jump-host."}
+                      </p>
+                    ) : null}
+                  </div>
                 </div>
+                <Frame dense className="w-full flex flex-col">
+                  <FramePanel className="flex-1 px-5 pb-5 pt-5">
+                    <FlowAnalyticsDetail
+                      card={selFlowCard}
+                      analytics={displayedFlow}
+                      range={range}
+                      onRange={(r) => setRange(r as Range)}
+                      selectedIface={flowIface}
+                      onIface={setFlowIface}
+                      liveHint={displayedFlow?.live ? "live" : undefined}
+                      emptyHint={flowEmptyHint(flowStats)}
+                    />
+                  </FramePanel>
+                </Frame>
               </div>
-              {liveError && (
-                <div className="text-xs text-destructive bg-destructive/10 border border-destructive/20 rounded-md px-3 py-2">
-                  {liveError}
-                </div>
-              )}
-              <DataPageCard>
-                <TrafficFlowsDataGrid rows={flowStats?.talkers ?? []} />
-              </DataPageCard>
               <FlowOverlaySheet
                 open={overlayOpen}
                 onOpenChange={setOverlayOpen}

backend/package.json

@@ -14,7 +14,7 @@
     "test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts",
     "test:wireguard": "npx tsx src/services/wireguard-config.test.ts",
     "test:traffic-rate": "tsx src/services/traffic-rate.test.ts",
-    "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts",
+    "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-analytics.test.ts",
     "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts"
   },
   "dependencies": {

backend/src/db/index.ts

@@ -158,7 +158,7 @@ CREATE TABLE IF NOT EXISTS flow_buckets (
   FOREIGN KEY (server_id) REFERENCES servers(id) ON DELETE CASCADE
 );
 CREATE UNIQUE INDEX IF NOT EXISTS idx_flow_buckets_unique
-  ON flow_buckets(server_id, bucket_at, src, dst, proto, src_port, dst_port);
+  ON flow_buckets(server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface);
 CREATE INDEX IF NOT EXISTS idx_flow_buckets_server_time
   ON flow_buckets(server_id, bucket_at);
 
@@ -805,6 +805,29 @@ SELECT 1, 'https://acme-v02.api.letsencrypt.org/directory', '', '', ''
 WHERE NOT EXISTS (SELECT 1 FROM acme_settings WHERE id = 1);
 `)
 
+{
+  const flowIndexes = sqlite.prepare(`PRAGMA index_list('flow_buckets')`).all() as Array<{
+    name?: string
+    unique?: number
+  }>
+  let hasIfaceUnique = false
+  for (const idx of flowIndexes) {
+    if (!idx.name || !idx.unique) continue
+    const info = sqlite.prepare(`PRAGMA index_info(${JSON.stringify(idx.name)})`).all() as Array<{ name?: string }>
+    const names = info.map((c) => c.name)
+    if (names.includes("in_iface") && names.includes("src") && names.includes("dst")) {
+      hasIfaceUnique = true
+    }
+  }
+  if (!hasIfaceUnique) {
+    sqlite.exec(`DROP INDEX IF EXISTS idx_flow_buckets_unique`)
+    sqlite.exec(`
+      CREATE UNIQUE INDEX IF NOT EXISTS idx_flow_buckets_unique
+        ON flow_buckets(server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
+    `)
+  }
+}
+
 const certIssueJobCols = sqlite.prepare(`PRAGMA table_info('certificate_issue_jobs')`).all() as Array<{ name?: string }>
 if (!certIssueJobCols.some((c) => c.name === "source")) {
   sqlite.exec(`ALTER TABLE certificate_issue_jobs ADD COLUMN source TEXT NOT NULL DEFAULT 'manual'`)

backend/src/db/schema.ts

@@ -198,7 +198,7 @@ export const flowBuckets = sqliteTable("flow_buckets", {
   inIface:   text("in_iface").notNull().default(""),
 }, (t) => [
   uniqueIndex("idx_flow_buckets_unique").on(
-    t.serverId, t.bucketAt, t.src, t.dst, t.proto, t.srcPort, t.dstPort,
+    t.serverId, t.bucketAt, t.src, t.dst, t.proto, t.srcPort, t.dstPort, t.inIface,
   ),
 ])
 

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

backend/src/services/traffic-collector.ts

@@ -5,9 +5,12 @@ import type { TrafficRunSnapshot } from "../types/scheduler-run-snapshot.js"
 import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
 import { MikrotikClient } from "./mikrotik.js"
 import { bpsToMbps, rateBpsFromDelta, shouldIncludeIface } from "./traffic-rate.js"
+import { rememberServerIfaces } from "./traffic-flow-ifindex.js"
 
 interface RosIfaceTraffic {
+  ".id"?: string
   name?: string
+  ifindex?: string
   running?: string
   disabled?: string
   "rx-byte"?: string
@@ -139,6 +142,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
       try {
         const client = MikrotikClient.fromServer(srv)
         const ifaces = await client.get<RosIfaceTraffic[]>("/interface")
+        rememberServerIfaces(srv.id, ifaces)
         const prevWave = readPreviousWave(srv.id)
         const nowMs = Date.parse(now)
         let sumRxMbps = 0

backend/src/services/traffic-flow-analytics.test.ts

@@ -0,0 +1,68 @@
+import assert from "node:assert/strict"
+import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js"
+import {
+  ingestParsedFlowsForServerForTests,
+  resetFlowRingsForTests,
+} from "./traffic-flow-ingest.js"
+import { buildFlowAnalytics } from "./traffic-flow-analytics.js"
+
+resetIfaceCacheForTests()
+resetFlowRingsForTests()
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "ether1" },
+  { ".id": "*A", name: "wg-flow" },
+])
+
+ingestParsedFlowsForServerForTests(7, [
+  {
+    src: "10.1.1.8",
+    dst: "8.8.8.8",
+    proto: 6,
+    srcPort: 51234,
+    dstPort: 443,
+    bytes: 12_000,
+    packets: 10,
+    inIface: "2",
+    outIface: "10",
+  },
+  {
+    src: "10.1.1.8",
+    dst: "1.1.1.1",
+    proto: 17,
+    srcPort: 53000,
+    dstPort: 53,
+    bytes: 800,
+    packets: 4,
+    inIface: "2",
+    outIface: "",
+  },
+])
+
+try {
+  const all = buildFlowAnalytics({ minutes: 5, serverId: 7 })
+  assert.equal(all.applications[0]?.label, "HTTPS")
+  assert.ok(all.protocols.some((p) => p.label === "TCP"))
+  assert.equal(all.ifaces[0]?.name, "ether1")
+  assert.notEqual(all.ifaces[0]?.name, "2")
+  const conv = all.conversationsList[0]
+  assert.ok(conv)
+  assert.equal(conv.inIface, "ether1")
+  assert.equal(conv.inIfaceIndex, "2")
+  assert.equal(conv.application, "HTTPS")
+  assert.ok(!/^\d+$/.test(conv.inIface))
+
+  const filtered = buildFlowAnalytics({ minutes: 5, serverId: 7, iface: "ether1" })
+  assert.ok(filtered.bytes >= 12_000)
+  assert.equal(filtered.ifaces[0]?.name, "ether1")
+
+  const miss = buildFlowAnalytics({ minutes: 5, serverId: 7, iface: "wg-flow" })
+  assert.equal(miss.conversations, 0)
+
+  const other = buildFlowAnalytics({ minutes: 5, serverId: 99 })
+  assert.equal(other.conversations, 0)
+} finally {
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+}
+
+console.log("traffic-flow-analytics.test.ts: ok")

backend/src/services/traffic-flow-analytics.ts

@@ -0,0 +1,303 @@
+import { eq } from "drizzle-orm"
+import { db } from "../db/index.js"
+import { appUsers, servers, userInterfaceBindings } from "../db/schema.js"
+import type {
+  FlowAnalyticsDto,
+  FlowBreakdownRow,
+  FlowClientsDto,
+  FlowEntityCard,
+  FlowExportersDto,
+  FlowTalkerDto,
+} from "@mmapp/contracts/traffic-flow"
+import { protoName } from "./traffic-flow-parse.js"
+import {
+  getFlowListenerState,
+  getRingMbps,
+  listStoredFlowRows,
+  type PendingFlowRow,
+} from "./traffic-flow-ingest.js"
+import { resolveIfaceName } from "./traffic-flow-ifaces.js"
+import { getTrafficFlowSettingsRow, listHostPeers } from "./traffic-flow-settings.js"
+import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
+
+export interface FlowAnalyticsQuery {
+  minutes: number
+  serverId?: number
+  userId?: string
+  iface?: string
+}
+
+function bpsToMbps(bps: number): number {
+  return bps / 1_000_000
+}
+
+function topN(map: Map<string, { bytes: number; packets: number }>, windowSec: number, n: number): FlowBreakdownRow[] {
+  const total = [...map.values()].reduce((a, v) => a + v.bytes, 0) || 1
+  return [...map.entries()]
+    .sort((a, b) => b[1].bytes - a[1].bytes)
+    .slice(0, n)
+    .map(([id, v]) => ({
+      id,
+      label: id,
+      bytes: v.bytes,
+      packets: v.packets,
+      bps: (v.bytes * 8) / windowSec,
+      percent: (v.bytes / total) * 100,
+    }))
+}
+
+function bump(map: Map<string, { bytes: number; packets: number }>, id: string, bytes: number, packets: number) {
+  const prev = map.get(id) ?? { bytes: 0, packets: 0 }
+  prev.bytes += bytes
+  prev.packets += packets
+  map.set(id, prev)
+}
+
+function userIfaceAllow(userId: string): Map<number, Set<string>> | null {
+  if (!userId) return null
+  const binds = db.select().from(userInterfaceBindings).where(eq(userInterfaceBindings.userId, userId)).all()
+  const allow = new Map<number, Set<string>>()
+  for (const b of binds) {
+    const set = allow.get(b.serverId) ?? new Set<string>()
+    set.add(b.interfaceName)
+    allow.set(b.serverId, set)
+  }
+  return allow
+}
+
+function seriesFromRows(rows: PendingFlowRow[], minutes: number): { rx: number[]; tx: number[] } {
+  const slots = Math.min(60, Math.max(5, minutes))
+  const slotMs = (minutes * 60_000) / slots
+  const start = Date.now() - minutes * 60_000
+  const rx = Array(slots).fill(0) as number[]
+  const tx = Array(slots).fill(0) as number[]
+  for (const r of rows) {
+    const t = Date.parse(r.bucketAt)
+    if (!Number.isFinite(t)) continue
+    const idx = Math.min(slots - 1, Math.max(0, Math.floor((t - start) / slotMs)))
+    rx[idx] += r.bytes
+  }
+  const slotSec = Math.max(1, slotMs / 1000)
+  return {
+    rx: rx.map((b) => bpsToMbps((b * 8) / slotSec)),
+    tx,
+  }
+}
+
+function snapshotStatus(serverId: number): FlowEntityCard["status"] {
+  void serverId
+  return "online"
+}
+
+export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
+  const settings = getTrafficFlowSettingsRow()
+  const top = Math.min(50, Math.max(10, settings.topN))
+  const windowSec = Math.max(60, q.minutes * 60)
+  const sinceIso = new Date(Date.now() - q.minutes * 60_000).toISOString()
+  const raw = listStoredFlowRows(sinceIso)
+  const allow = q.userId ? userIfaceAllow(q.userId) : null
+  const serverRows = db.select().from(servers).all()
+  const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host]))
+
+  const applications = new Map<string, { bytes: number; packets: number }>()
+  const protocols = new Map<string, { bytes: number; packets: number }>()
+  const sources = new Map<string, { bytes: number; packets: number }>()
+  const destinations = new Map<string, { bytes: number; packets: number }>()
+  const ifacesMap = new Map<string, { bytes: number; packets: number; index: string }>()
+  const conv = new Map<string, FlowTalkerDto & { rawBytes: number }>()
+  const srcs = new Set<string>()
+  const dsts = new Set<string>()
+  let totalBytes = 0
+  let totalPackets = 0
+  const matched: PendingFlowRow[] = []
+
+  for (const r of raw) {
+    const resolved = resolveIfaceName(r.serverId, r.inIface)
+    if (!flowRowMatchesFilter(r, resolved.name, q, allow)) continue
+    matched.push(r)
+    totalBytes += r.bytes
+    totalPackets += r.packets
+    srcs.add(r.src)
+    dsts.add(r.dst)
+    const app = applicationName(r.proto, r.dstPort, r.srcPort)
+    bump(applications, app, r.bytes, r.packets)
+    bump(protocols, protoName(r.proto), r.bytes, r.packets)
+    bump(sources, r.src, r.bytes, r.packets)
+    bump(destinations, r.dst, r.bytes, r.packets)
+    const ifaceKey = resolved.name
+    const prevIf = ifacesMap.get(ifaceKey) ?? { bytes: 0, packets: 0, index: resolved.index }
+    prevIf.bytes += r.bytes
+    prevIf.packets += r.packets
+    ifacesMap.set(ifaceKey, prevIf)
+
+    const ckey = `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}|${r.inIface}`
+    const prev = conv.get(ckey)
+    if (prev) {
+      prev.rawBytes += r.bytes
+      prev.bytes += r.bytes
+      prev.packets += r.packets
+    } else {
+      conv.set(ckey, {
+        serverId: String(r.serverId),
+        serverName: nameById.get(r.serverId) ?? String(r.serverId),
+        src: r.src,
+        dst: r.dst,
+        proto: r.proto,
+        protoName: protoName(r.proto),
+        srcPort: r.srcPort,
+        dstPort: r.dstPort,
+        bytes: r.bytes,
+        packets: r.packets,
+        bps: 0,
+        inIface: resolved.name,
+        inIfaceIndex: resolved.index,
+        application: app,
+        rawBytes: r.bytes,
+      })
+    }
+  }
+
+  const conversationsList = [...conv.values()]
+    .map((t) => ({ ...t, bps: (t.rawBytes * 8) / windowSec }))
+    .sort((a, b) => b.bytes - a.bytes)
+    .slice(0, top)
+    .map(({ rawBytes: _raw, ...rest }) => rest)
+
+  let topProto = "—"
+  let topProtoBytes = 0
+  for (const [label, v] of protocols) {
+    if (v.bytes > topProtoBytes) {
+      topProtoBytes = v.bytes
+      topProto = label
+    }
+  }
+
+  const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : "__all__"
+  const ringServer = q.serverId ?? (matched[0]?.serverId ?? 0)
+  const ring = ringServer
+    ? getRingMbps(ringServer, ifaceFilter === "__all__" ? "__all__" : (ifacesMap.get(ifaceFilter)?.index || ifaceFilter))
+    : { rx: Array(60).fill(0) as number[], tx: Array(60).fill(0) as number[], rxNow: 0, txNow: 0 }
+
+  const fromBuckets = seriesFromRows(matched, q.minutes)
+  const rxSeries = q.minutes <= 15 ? ring.rx : fromBuckets.rx
+  const txSeries = q.minutes <= 15 ? ring.tx : fromBuckets.tx
+
+  const ifaceRows = [...ifacesMap.entries()]
+    .sort((a, b) => b[1].bytes - a[1].bytes)
+    .map(([name, v]) => ({
+      name,
+      index: v.index,
+      bps: (v.bytes * 8) / windowSec,
+    }))
+
+  const protoBreakdown = topN(protocols, windowSec, top)
+  const listener = getFlowListenerState()
+
+  return {
+    bpsNow: (ring.rxNow + ring.txNow) * 1_000_000 || (totalBytes * 8) / windowSec,
+    bytes: totalBytes,
+    packets: totalPackets,
+    conversations: conv.size,
+    uniqueSrc: srcs.size,
+    uniqueDst: dsts.size,
+    topProto,
+    rxSeries,
+    txSeries,
+    applications: topN(applications, windowSec, top),
+    protocols: protoBreakdown,
+    sources: topN(sources, windowSec, top),
+    destinations: topN(destinations, windowSec, top),
+    interfaces: [...ifacesMap.entries()].map(([label, v]) => ({
+      id: label,
+      label,
+      bytes: v.bytes,
+      packets: v.packets,
+      bps: (v.bytes * 8) / windowSec,
+      percent: totalBytes > 0 ? (v.bytes / totalBytes) * 100 : 0,
+    })).sort((a, b) => b.bytes - a.bytes),
+    conversationsList,
+    ifaces: ifaceRows,
+    live: listener.bound,
+  }
+}
+
+function cardFromServer(
+  s: typeof servers.$inferSelect,
+  minutes: number,
+): FlowEntityCard {
+  const analytics = buildFlowAnalytics({ minutes, serverId: s.id })
+  const ring = getRingMbps(s.id, "__all__")
+  return {
+    id: String(s.id),
+    name: s.name || s.host,
+    subtitle: s.host,
+    site: s.site || "—",
+    country: s.country || "UN",
+    status: snapshotStatus(s.id),
+    rxNow: ring.rxNow || bpsToMbps(analytics.bpsNow),
+    txNow: ring.txNow,
+    sessions: analytics.conversations,
+    rxSeries: ring.rx.some((v) => v > 0) ? ring.rx : analytics.rxSeries,
+    txSeries: ring.tx,
+    bytes: analytics.bytes,
+  }
+}
+
+export function listFlowExporters(minutes: number): FlowExportersDto {
+  const settings = getTrafficFlowSettingsRow()
+  const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString()
+  const rows = listStoredFlowRows(sinceIso)
+  const ids = new Set<number>()
+  for (const r of rows) ids.add(r.serverId)
+  for (const p of listHostPeers()) ids.add(p.serverId)
+  const serverRows = db.select().from(servers).all()
+  const exporters = serverRows
+    .filter((s) => ids.has(s.id))
+    .map((s) => cardFromServer(s, minutes))
+    .sort((a, b) => b.rxNow - a.rxNow)
+  const listener = getFlowListenerState()
+  return {
+    exporters,
+    lastExporterIp: settings.lastExporterIp ?? null,
+    lastError: settings.lastError || null,
+    packetsReceived: settings.packetsReceived,
+    lastDatagramAt: settings.lastDatagramAt ?? null,
+    listenerBound: listener.bound,
+    listenerAddress: listener.address,
+  }
+}
+
+export function listFlowClients(minutes: number): FlowClientsDto {
+  const users = db.select().from(appUsers).all()
+  const binds = db.select().from(userInterfaceBindings).all()
+  const byUser = new Map<string, typeof binds>()
+  for (const b of binds) {
+    const list = byUser.get(b.userId) ?? []
+    list.push(b)
+    byUser.set(b.userId, list)
+  }
+  const clients: FlowEntityCard[] = []
+  for (const u of users) {
+    const userBinds = byUser.get(u.id) ?? []
+    if (userBinds.length === 0) continue
+    const analytics = buildFlowAnalytics({ minutes, userId: u.id })
+    const firstServer = userBinds[0]?.serverId
+    const ring = firstServer ? getRingMbps(firstServer, "__all__") : { rx: Array(60).fill(0) as number[], tx: Array(60).fill(0) as number[], rxNow: 0, txNow: 0 }
+    clients.push({
+      id: u.id,
+      name: u.login,
+      subtitle: u.name || u.login,
+      site: `${userBinds.length} ifaces`,
+      country: "UN",
+      status: u.active ? "online" : "offline",
+      rxNow: bpsToMbps(analytics.bpsNow) || ring.rxNow,
+      txNow: ring.txNow,
+      sessions: analytics.conversations,
+      rxSeries: analytics.rxSeries,
+      txSeries: analytics.txSeries,
+      bytes: analytics.bytes,
+    })
+  }
+  clients.sort((a, b) => b.rxNow - a.rxNow)
+  return { clients }
+}

backend/src/services/traffic-flow-apps.ts

@@ -0,0 +1,79 @@
+import { protoName } from "./traffic-flow-parse.js"
+
+const WELL_KNOWN: Record<string, string> = {
+  "6:80": "HTTP",
+  "6:443": "HTTPS",
+  "6:8080": "HTTP-alt",
+  "6:8443": "HTTPS-alt",
+  "6:22": "SSH",
+  "6:21": "FTP",
+  "6:25": "SMTP",
+  "6:110": "POP3",
+  "6:143": "IMAP",
+  "6:993": "IMAPS",
+  "6:995": "POP3S",
+  "6:587": "SMTP",
+  "6:465": "SMTPS",
+  "6:3306": "MySQL",
+  "6:5432": "PostgreSQL",
+  "6:6379": "Redis",
+  "6:3389": "RDP",
+  "6:445": "SMB",
+  "6:139": "NetBIOS",
+  "6:179": "BGP",
+  "6:8291": "WinBox",
+  "6:8728": "ROS-API",
+  "6:8729": "ROS-API-SSL",
+  "17:53": "DNS",
+  "6:53": "DNS",
+  "17:123": "NTP",
+  "17:161": "SNMP",
+  "17:162": "SNMP-trap",
+  "17:500": "IKE",
+  "17:4500": "NAT-T",
+  "17:1194": "OpenVPN",
+  "17:51820": "WireGuard",
+  "17:4789": "VXLAN",
+  "17:4739": "IPFIX",
+  "17:2055": "NetFlow",
+  "17:67": "DHCP",
+  "17:68": "DHCP",
+  "17:69": "TFTP",
+  "17:1812": "RADIUS",
+  "1:0": "ICMP",
+  "47:0": "GRE",
+  "50:0": "ESP",
+  "89:0": "OSPF",
+}
+
+export function applicationName(proto: number, dstPort: number, srcPort = 0): string {
+  if (proto === 1) return "ICMP"
+  if (proto === 47) return "GRE"
+  if (proto === 50) return "ESP"
+  if (proto === 89) return "OSPF"
+  const dstKey = `${proto}:${dstPort}`
+  const srcKey = `${proto}:${srcPort}`
+  return WELL_KNOWN[dstKey] ?? WELL_KNOWN[srcKey] ?? `${protoName(proto)}/${dstPort || srcPort || "—"}`
+}
+
+export interface FlowMatchQuery {
+  serverId?: number
+  userId?: string
+  iface?: string
+}
+
+export function flowRowMatchesFilter(
+  row: { serverId: number; inIface: string },
+  resolvedName: string,
+  q: FlowMatchQuery,
+  allow: Map<number, Set<string>> | null,
+): boolean {
+  if (q.serverId != null && row.serverId !== q.serverId) return false
+  if (allow) {
+    const names = allow.get(row.serverId)
+    if (!names || !names.has(resolvedName)) return false
+  }
+  const iface = q.iface && q.iface !== "__all__" ? q.iface : ""
+  if (iface && resolvedName !== iface && row.inIface !== iface) return false
+  return true
+}

backend/src/services/traffic-flow-host-files.ts

@@ -29,7 +29,8 @@ export function buildHostComposeOverride(): string {
     "# Docker Compose merge для /opt/cdn-mm",
     "# Не править docker-compose.yml. Traefik не трогать.",
     "# Сначала: wg-quick up wg-flow  (адрес " + row.collectorIp + ")",
-    "# затем: docker compose up -d backend",
+    "# затем: docker compose up -d --force-recreate backend",
+    "# Docker userland-proxy может SNAT UDP source в 172.x — ingest сопоставит единственный JH.",
     "",
     "services:",
     "  backend:",
@@ -87,19 +88,17 @@ cat > "\$COMPOSE_DIR/docker-compose.override.yml" <<'OVEOF'
 ${override}OVEOF
 
 cd "\$COMPOSE_DIR"
-docker compose up -d backend
+docker compose up -d --force-recreate backend
 
-echo "=== UDP \${FLOW_PORT} на хосте ==="
+echo "=== UDP \${FLOW_PORT} на хосте (ожидаем \${COLLECTOR_IP}:\${FLOW_PORT} docker-proxy) ==="
 ss -ulnp | grep -E "\${FLOW_PORT}" || true
 echo "=== PortBindings mmapp-backend ==="
 docker inspect -f '{{json .HostConfig.PortBindings}}' mmapp-backend
 echo "=== handshake (keepalive 25s к JH:13232) ==="
 wg show wg-flow
 
-# ufw: исходящий WG не открывать; 4739 на WAN не публиковать
-if command -v ufw >/dev/null 2>&1; then
-  ufw deny "\${FLOW_PORT}/udp" comment 'ipfix-not-public' || true
-fi
+# nft на хосте MM не трогаем. Bind только на COLLECTOR_IP, не 0.0.0.0.
+# Если backend стартовал до wg-flow: docker compose up -d --force-recreate backend
 
 echo "Готово. Traefik не трогали. UDP \${FLOW_PORT} только на \${COLLECTOR_IP}, не на 0.0.0.0."
 `

backend/src/services/traffic-flow-ifaces.test.ts

@@ -0,0 +1,43 @@
+import assert from "node:assert/strict"
+import {
+  rememberServerIfaces,
+  resetIfaceCacheForTests,
+  resolveIfaceName,
+  rosIdToIfIndex,
+} from "./traffic-flow-ifindex.js"
+import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
+
+assert.equal(rosIdToIfIndex("*A"), 10)
+assert.equal(rosIdToIfIndex("*D"), 13)
+assert.equal(rosIdToIfIndex("*2"), 2)
+assert.equal(rosIdToIfIndex("*9"), 9)
+assert.equal(rosIdToIfIndex("0"), 0)
+assert.equal(rosIdToIfIndex(""), null)
+
+resetIfaceCacheForTests()
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "ether1" },
+  { ".id": "*A", name: "wg-flow" },
+  { ".id": "*D", name: "bridge" },
+])
+assert.equal(resolveIfaceName(7, "2").name, "ether1")
+assert.equal(resolveIfaceName(7, "10").name, "wg-flow")
+assert.equal(resolveIfaceName(7, "13").name, "bridge")
+assert.equal(resolveIfaceName(7, "0").name, "—")
+assert.equal(resolveIfaceName(7, "ether1").name, "ether1")
+assert.equal(resolveIfaceName(7, "99").name, "#99")
+
+assert.equal(applicationName(6, 443), "HTTPS")
+assert.equal(applicationName(17, 53), "DNS")
+assert.equal(applicationName(6, 22), "SSH")
+assert.equal(applicationName(17, 51820), "WireGuard")
+assert.equal(applicationName(6, 179), "BGP")
+
+const allow = new Map<number, Set<string>>([[7, new Set(["ether1", "wg-flow"])]])
+assert.equal(flowRowMatchesFilter({ serverId: 7, inIface: "2" }, "ether1", {}, allow), true)
+assert.equal(flowRowMatchesFilter({ serverId: 7, inIface: "2" }, "bridge", {}, allow), false)
+assert.equal(flowRowMatchesFilter({ serverId: 7, inIface: "2" }, "ether1", { iface: "ether1" }, allow), true)
+assert.equal(flowRowMatchesFilter({ serverId: 7, inIface: "2" }, "ether1", { iface: "wg-flow" }, allow), false)
+assert.equal(flowRowMatchesFilter({ serverId: 8, inIface: "2" }, "ether1", { serverId: 7 }, null), false)
+
+console.log("traffic-flow-ifaces.test.ts: ok")

backend/src/services/traffic-flow-ifaces.ts

@@ -0,0 +1,36 @@
+import { eq } from "drizzle-orm"
+import { db } from "../db/index.js"
+import { servers } from "../db/schema.js"
+import { MikrotikClient } from "./mikrotik.js"
+import {
+  ifaceCacheFresh,
+  rememberServerIfaces,
+  type RosIfaceIndexRow,
+} from "./traffic-flow-ifindex.js"
+
+export {
+  ifaceCacheHas,
+  rememberServerIfaces,
+  resetIfaceCacheForTests,
+  resolveIfaceName,
+  rosIdToIfIndex,
+} from "./traffic-flow-ifindex.js"
+
+const inflight = new Set<number>()
+
+export async function refreshServerIfaces(serverId: number, force = false): Promise<void> {
+  if (inflight.has(serverId)) return
+  if (!force && ifaceCacheFresh(serverId)) return
+  inflight.add(serverId)
+  try {
+    const row = db.select().from(servers).where(eq(servers.id, serverId)).limit(1).all()[0]
+    if (!row) return
+    const client = MikrotikClient.fromServer(row)
+    const ifaces = await client.get<RosIfaceIndexRow[]>("/interface")
+    rememberServerIfaces(serverId, Array.isArray(ifaces) ? ifaces : [])
+  } catch {
+    /* keep previous cache */
+  } finally {
+    inflight.delete(serverId)
+  }
+}

backend/src/services/traffic-flow-ifindex.ts

@@ -0,0 +1,59 @@
+export interface RosIfaceIndexRow {
+  ".id"?: string
+  name?: string
+  ifindex?: string
+}
+
+const cache = new Map<number, Map<number, string>>()
+const fetchedAt = new Map<number, number>()
+
+export const IFACE_CACHE_TTL_MS = 60_000
+
+/** RouterOS `.id` (`*A`) → SNMP ifIndex (10). */
+export function rosIdToIfIndex(id: string | undefined | null): number | null {
+  if (!id) return null
+  const raw = String(id).trim()
+  const hex = raw.startsWith("*") ? raw.slice(1) : raw
+  if (!hex || !/^[0-9a-fA-F]+$/.test(hex)) return null
+  const n = parseInt(hex, 16)
+  return Number.isFinite(n) ? n : null
+}
+
+export function rememberServerIfaces(serverId: number, rows: RosIfaceIndexRow[]): void {
+  const map = new Map<number, string>()
+  for (const row of rows) {
+    const name = String(row.name ?? "").trim()
+    if (!name) continue
+    const fromProp = Number.parseInt(String(row.ifindex ?? ""), 10)
+    const idx = Number.isFinite(fromProp) && fromProp > 0
+      ? fromProp
+      : rosIdToIfIndex(row[".id"])
+    if (idx != null && idx > 0) map.set(idx, name)
+  }
+  cache.set(serverId, map)
+  fetchedAt.set(serverId, Date.now())
+}
+
+export function resolveIfaceName(serverId: number, indexOrName: string): { name: string; index: string } {
+  const trimmed = String(indexOrName ?? "").trim()
+  if (!trimmed || trimmed === "0") return { name: "—", index: trimmed }
+  if (!/^\d+$/.test(trimmed)) return { name: trimmed, index: "" }
+  const idx = Number(trimmed)
+  const name = cache.get(serverId)?.get(idx)
+  if (name) return { name, index: trimmed }
+  return { name: `#${trimmed}`, index: trimmed }
+}
+
+export function ifaceCacheHas(serverId: number): boolean {
+  return cache.has(serverId)
+}
+
+export function ifaceCacheFresh(serverId: number, ttlMs = IFACE_CACHE_TTL_MS): boolean {
+  const prev = fetchedAt.get(serverId) ?? 0
+  return Boolean(prev && Date.now() - prev < ttlMs && cache.has(serverId))
+}
+
+export function resetIfaceCacheForTests(): void {
+  cache.clear()
+  fetchedAt.clear()
+}

backend/src/services/traffic-flow-ingest.ts

@@ -4,17 +4,38 @@ import { db } from "../db/index.js"
 import { flowBuckets, servers } from "../db/schema.js"
 import type { FlowStatsDto, FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
 import { parseFlowPacket, protoName, type ParsedFlow } from "./traffic-flow-parse.js"
+import { pickServerIdForExporter } from "./traffic-flow-map-exporter.js"
 import {
   getTrafficFlowSettingsRow,
+  listHostPeers,
   recordFlowListenerError,
   recordFlowPacket,
 } from "./traffic-flow-settings.js"
+import { ifaceCacheHas, refreshServerIfaces, resolveIfaceName } from "./traffic-flow-ifaces.js"
+import { applicationName } from "./traffic-flow-apps.js"
 
 export interface FlowListenerState {
   bound: boolean
   address: string | null
 }
 
+export interface PendingFlowRow {
+  serverId: number
+  bucketAt: string
+  src: string
+  dst: string
+  proto: number
+  srcPort: number
+  dstPort: number
+  bytes: number
+  packets: number
+  inIface: string
+  outIface: string
+}
+
+const TICK_MS = 2_000
+const RING_LEN = 60
+
 let socket: Socket | null = null
 let state: FlowListenerState = { bound: false, address: null }
 const pending = new Map<string, {
@@ -26,6 +47,9 @@ const pending = new Map<string, {
 }>()
 let flushTimer: ReturnType<typeof setInterval> | null = null
 
+const tickAccum = new Map<string, { inBytes: number; outBytes: number }>()
+const rings = new Map<string, { inBps: number[]; outBps: number[] }>()
+
 export function getFlowListenerState(): FlowListenerState {
   return state
 }
@@ -36,17 +60,96 @@ function minuteBucketIso(at = Date.now()): string {
   return d.toISOString()
 }
 
-function resolveServerId(exporterIp: string): number | null {
-  const exact = db.select().from(servers).where(eq(servers.mgmtTunnelIp, exporterIp)).limit(1).all()[0]
-  return exact ? exact.id : null
+function ringKey(serverId: number, iface: string): string {
+  return `${serverId}\0${iface || "__all__"}`
 }
 
-function queueFlows(exporterIp: string, flows: ParsedFlow[]) {
+function bumpTick(key: string, inBytes: number, outBytes: number): void {
+  const prev = tickAccum.get(key) ?? { inBytes: 0, outBytes: 0 }
+  prev.inBytes += inBytes
+  prev.outBytes += outBytes
+  tickAccum.set(key, prev)
+}
+
+function addToTick(serverId: number, inIface: string, outIface: string, bytes: number): void {
+  bumpTick(ringKey(serverId, "__all__"), bytes, 0)
+  if (inIface) bumpTick(ringKey(serverId, inIface), bytes, 0)
+  if (outIface && outIface !== inIface) bumpTick(ringKey(serverId, outIface), 0, bytes)
+}
+
+function emptyRing(): { inBps: number[]; outBps: number[] } {
+  return { inBps: Array(RING_LEN).fill(0), outBps: Array(RING_LEN).fill(0) }
+}
+
+export function rollFlowRings(): void {
+  const keys = new Set([...tickAccum.keys(), ...rings.keys()])
+  const sec = TICK_MS / 1000
+  for (const key of keys) {
+    const acc = tickAccum.get(key) ?? { inBytes: 0, outBytes: 0 }
+    tickAccum.delete(key)
+    const inBps = (acc.inBytes * 8) / sec
+    const outBps = (acc.outBytes * 8) / sec
+    let ring = rings.get(key)
+    if (!ring) {
+      ring = emptyRing()
+      rings.set(key, ring)
+    }
+    ring.inBps.push(inBps)
+    ring.inBps.shift()
+    ring.outBps.push(outBps)
+    ring.outBps.shift()
+  }
+}
+
+export function getRingMbps(serverId: number, iface = "__all__"): {
+  rx: number[]
+  tx: number[]
+  rxNow: number
+  txNow: number
+} {
+  const ring = rings.get(ringKey(serverId, iface))
+  const scale = 1_000_000
+  if (!ring) {
+    return { rx: Array(RING_LEN).fill(0), tx: Array(RING_LEN).fill(0), rxNow: 0, txNow: 0 }
+  }
+  return {
+    rx: ring.inBps.map((b) => b / scale),
+    tx: ring.outBps.map((b) => b / scale),
+    rxNow: (ring.inBps[RING_LEN - 1] ?? 0) / scale,
+    txNow: (ring.outBps[RING_LEN - 1] ?? 0) / scale,
+  }
+}
+
+function resolveServerId(exporterIp: string): number | null {
+  const settings = getTrafficFlowSettingsRow()
+  const rows = db.select({
+    id: servers.id,
+    host: servers.host,
+    mgmtTunnelIp: servers.mgmtTunnelIp,
+  }).from(servers).all()
+  const byTunnelIp = new Map<string, number>()
+  const hostIps = new Map<string, number>()
+  for (const row of rows) {
+    if (row.mgmtTunnelIp) byTunnelIp.set(row.mgmtTunnelIp, row.id)
+    if (/^\d{1,3}(?:\.\d{1,3}){3}$/.test(row.host)) hostIps.set(row.host, row.id)
+  }
+  return pickServerIdForExporter({
+    exporterIp,
+    overlayPrefix: settings.prefix,
+    byTunnelIp,
+    peers: listHostPeers(),
+    hostIps,
+  })
+}
+
+function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean {
   const serverId = resolveServerId(exporterIp)
-  if (serverId == null) return
+  if (serverId == null) return false
+  if (!ifaceCacheHas(serverId)) void refreshServerIfaces(serverId)
   const bucketAt = minuteBucketIso()
   for (const flow of flows) {
-    const key = `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}`
+    addToTick(serverId, flow.inIface, flow.outIface, flow.bytes)
+    const key = `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}\0${flow.inIface}`
     const prev = pending.get(key)
     if (prev) {
       prev.bytes += flow.bytes
@@ -61,6 +164,23 @@ function queueFlows(exporterIp: string, flows: ParsedFlow[]) {
       })
     }
   }
+  return true
+}
+
+export function peekPendingFlows(): PendingFlowRow[] {
+  return [...pending.values()].map((row) => ({
+    serverId: row.serverId,
+    bucketAt: row.bucketAt,
+    src: row.flow.src || "0.0.0.0",
+    dst: row.flow.dst || "0.0.0.0",
+    proto: row.flow.proto,
+    srcPort: row.flow.srcPort,
+    dstPort: row.flow.dstPort,
+    bytes: row.bytes,
+    packets: row.packets,
+    inIface: row.flow.inIface,
+    outIface: row.flow.outIface,
+  }))
 }
 
 function flushPending() {
@@ -93,6 +213,7 @@ function flushPending() {
           flowBuckets.proto,
           flowBuckets.srcPort,
           flowBuckets.dstPort,
+          flowBuckets.inIface,
         ],
         set: {
           bytes: sql`${flowBuckets.bytes} + excluded.bytes`,
@@ -125,11 +246,23 @@ function flushPending() {
   }
 }
 
+function onTick() {
+  rollFlowRings()
+  flushPending()
+}
+
 function onMessage(msg: Buffer, rinfo: { address: string }) {
   try {
     const flows = parseFlowPacket(msg, rinfo.address)
     recordFlowPacket(rinfo.address)
-    if (flows.length) queueFlows(rinfo.address, flows)
+    if (!flows.length) return
+    if (!queueFlows(rinfo.address, flows)) {
+      recordFlowListenerError(
+        `IPFIX от ${rinfo.address}: нет jump-host с адресом wg-flow. Docker SNAT (172.x) при нескольких JH не различим.`,
+      )
+      return
+    }
+    recordFlowListenerError("")
   } catch (e) {
     recordFlowListenerError(e instanceof Error ? e.message : String(e))
   }
@@ -168,12 +301,46 @@ export function startTrafficFlowListener() {
     recordFlowListenerError("")
   })
   socket = sock
-  flushTimer = setInterval(flushPending, 15_000)
+  flushTimer = setInterval(onTick, TICK_MS)
+}
+
+export function listStoredFlowRows(sinceIso: string): PendingFlowRow[] {
+  const stored = db.select().from(flowBuckets).where(gte(flowBuckets.bucketAt, sinceIso)).all()
+  const merged = new Map<string, PendingFlowRow>()
+  for (const r of stored) {
+    const key = `${r.serverId}|${r.bucketAt}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}|${r.inIface}`
+    merged.set(key, {
+      serverId: r.serverId,
+      bucketAt: r.bucketAt,
+      src: r.src,
+      dst: r.dst,
+      proto: r.proto,
+      srcPort: r.srcPort,
+      dstPort: r.dstPort,
+      bytes: r.bytes,
+      packets: r.packets,
+      inIface: r.inIface,
+      outIface: "",
+    })
+  }
+  for (const p of peekPendingFlows()) {
+    if (p.bucketAt < sinceIso) continue
+    const key = `${p.serverId}|${p.bucketAt}|${p.src}|${p.dst}|${p.proto}|${p.srcPort}|${p.dstPort}|${p.inIface}`
+    const prev = merged.get(key)
+    if (prev) {
+      prev.bytes += p.bytes
+      prev.packets += p.packets
+    } else {
+      merged.set(key, { ...p })
+    }
+  }
+  return [...merged.values()]
 }
 
 export function listFlowTalkers(minutes = 5): FlowStatsDto {
+  const settings = getTrafficFlowSettingsRow()
   const rangeStart = new Date(Date.now() - minutes * 60_000).toISOString()
-  const rows = db.select().from(flowBuckets).where(gte(flowBuckets.bucketAt, rangeStart)).all()
+  const rows = listStoredFlowRows(rangeStart)
   const serverRows = db.select().from(servers).all()
   const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host]))
   const agg = new Map<string, FlowTalkerDto & { rawBytes: number }>()
@@ -183,7 +350,8 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto {
   const exporters = new Set<number>()
   let totalBytes = 0
   for (const r of rows) {
-    const key = `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}`
+    const resolved = resolveIfaceName(r.serverId, r.inIface)
+    const key = `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}|${r.inIface}`
     const prev = agg.get(key)
     const bytes = r.bytes
     totalBytes += bytes
@@ -208,7 +376,9 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto {
         bytes,
         packets: r.packets,
         bps: 0,
-        inIface: r.inIface,
+        inIface: resolved.name,
+        inIfaceIndex: resolved.index,
+        application: applicationName(r.proto, r.dstPort, r.srcPort),
         rawBytes: bytes,
       })
     }
@@ -217,7 +387,7 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto {
   const talkers = [...agg.values()]
     .map((t) => ({ ...t, bps: (t.rawBytes * 8) / windowSec }))
     .sort((a, b) => b.bytes - a.bytes)
-    .slice(0, getTrafficFlowSettingsRow().topN)
+    .slice(0, settings.topN)
     .map(({ rawBytes: _raw, ...rest }) => rest)
   let topProto = "—"
   let topProtoBytes = 0
@@ -234,10 +404,46 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto {
     uniqueDst: dsts.size,
     topProto,
     talkers,
+    lastExporterIp: settings.lastExporterIp ?? null,
+    lastError: settings.lastError || null,
+    packetsReceived: settings.packetsReceived,
+    lastDatagramAt: settings.lastDatagramAt ?? null,
+    listenerBound: state.bound,
+    listenerAddress: state.address,
   }
 }
 
 export function ingestParsedFlowsForTests(exporterIp: string, flows: ParsedFlow[]) {
   queueFlows(exporterIp, flows)
+  rollFlowRings()
   flushPending()
 }
+
+/** Кладёт потоки в pending без flush в SQLite — для юнит-тестов аналитики. */
+export function ingestParsedFlowsForServerForTests(serverId: number, flows: ParsedFlow[]) {
+  const bucketAt = minuteBucketIso()
+  for (const flow of flows) {
+    addToTick(serverId, flow.inIface, flow.outIface, flow.bytes)
+    const key = `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}\0${flow.inIface}`
+    const prev = pending.get(key)
+    if (prev) {
+      prev.bytes += flow.bytes
+      prev.packets += flow.packets
+    } else {
+      pending.set(key, {
+        serverId,
+        bucketAt,
+        flow: { ...flow },
+        bytes: flow.bytes,
+        packets: flow.packets,
+      })
+    }
+  }
+  rollFlowRings()
+}
+
+export function resetFlowRingsForTests() {
+  tickAccum.clear()
+  rings.clear()
+  pending.clear()
+}

backend/src/services/traffic-flow-map-exporter.test.ts

@@ -0,0 +1,54 @@
+import assert from "node:assert/strict"
+import {
+  bareIpv4,
+  ipInCidr,
+  isNatMasqueradeExporter,
+  normalizeExporterIp,
+  pickServerIdForExporter,
+} from "./traffic-flow-map-exporter.js"
+
+assert.equal(normalizeExporterIp("::ffff:172.18.0.2"), "172.18.0.2")
+assert.equal(bareIpv4("10.255.254.3/32"), "10.255.254.3")
+assert.equal(ipInCidr("10.255.254.3", "10.255.254.0/24"), true)
+assert.equal(ipInCidr("172.18.0.2", "10.255.254.0/24"), false)
+assert.equal(isNatMasqueradeExporter("172.18.0.2", "10.255.254.0/24"), true)
+assert.equal(isNatMasqueradeExporter("10.255.254.3", "10.255.254.0/24"), false)
+assert.equal(isNatMasqueradeExporter("10.0.0.12", "10.255.254.0/24"), true)
+
+const byTunnel = new Map([["10.255.254.3", 7]])
+assert.equal(pickServerIdForExporter({
+  exporterIp: "10.255.254.3",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: byTunnel,
+  peers: [],
+  hostIps: new Map(),
+}), 7)
+
+assert.equal(pickServerIdForExporter({
+  exporterIp: "172.18.0.2",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: byTunnel,
+  peers: [{ serverId: 7, address: "10.255.254.3", allowedIps: ["10.255.254.3/32"] }],
+  hostIps: new Map(),
+}), 7)
+
+assert.equal(pickServerIdForExporter({
+  exporterIp: "172.18.0.2",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: new Map([["10.255.254.3", 7], ["10.255.254.4", 8]]),
+  peers: [
+    { serverId: 7, address: "10.255.254.3", allowedIps: ["10.255.254.3/32"] },
+    { serverId: 8, address: "10.255.254.4", allowedIps: ["10.255.254.4/32"] },
+  ],
+  hostIps: new Map(),
+}), null)
+
+assert.equal(pickServerIdForExporter({
+  exporterIp: "94.142.140.141",
+  overlayPrefix: "10.255.254.0/24",
+  byTunnelIp: byTunnel,
+  peers: [],
+  hostIps: new Map([["94.142.140.141", 7]]),
+}), 7)
+
+console.log("traffic-flow-map-exporter.test.ts: ok")

backend/src/services/traffic-flow-map-exporter.ts

@@ -0,0 +1,82 @@
+export interface OverlayPeerRef {
+  serverId: number
+  address: string
+  allowedIps: string[]
+}
+
+export function normalizeExporterIp(ip: string): string {
+  const trimmed = ip.trim()
+  if (trimmed.toLowerCase().startsWith("::ffff:")) return trimmed.slice(7)
+  return trimmed
+}
+
+export function bareIpv4(value: string): string {
+  const raw = normalizeExporterIp(value).split("/")[0]?.trim() ?? ""
+  return raw
+}
+
+function ipv4ToInt(ip: string): number | null {
+  const parts = ip.split(".")
+  if (parts.length !== 4) return null
+  const n = parts.map((x) => Number(x))
+  if (n.some((x) => !Number.isInteger(x) || x < 0 || x > 255)) return null
+  return ((n[0]! << 24) | (n[1]! << 16) | (n[2]! << 8) | n[3]!) >>> 0
+}
+
+export function ipInCidr(ip: string, cidr: string): boolean {
+  const host = bareIpv4(ip)
+  const [base, bitsRaw] = cidr.split("/")
+  const bits = Number(bitsRaw ?? 32)
+  const a = ipv4ToInt(host)
+  const b = ipv4ToInt(bareIpv4(base ?? ""))
+  if (a == null || b == null || !Number.isFinite(bits) || bits < 0 || bits > 32) return false
+  const mask = bits === 0 ? 0 : (0xffffffff << (32 - bits)) >>> 0
+  return (a & mask) === (b & mask)
+}
+
+/** Docker userland-proxy / bridge SNAT, не адрес из оверлея wg-flow. */
+export function isNatMasqueradeExporter(ip: string, overlayPrefix: string): boolean {
+  const host = bareIpv4(ip)
+  if (!host) return false
+  if (ipInCidr(host, overlayPrefix)) return false
+  return ipInCidr(host, "10.0.0.0/8")
+    || ipInCidr(host, "172.16.0.0/12")
+    || ipInCidr(host, "192.168.0.0/16")
+    || ipInCidr(host, "127.0.0.0/8")
+}
+
+export function pickServerIdForExporter(opts: {
+  exporterIp: string
+  overlayPrefix: string
+  byTunnelIp: Map<string, number>
+  peers: OverlayPeerRef[]
+  hostIps: Map<string, number>
+}): number | null {
+  const exporter = bareIpv4(opts.exporterIp)
+  if (!exporter) return null
+
+  const exact = opts.byTunnelIp.get(exporter)
+  if (exact != null) return exact
+
+  for (const [ip, id] of opts.byTunnelIp) {
+    if (bareIpv4(ip) === exporter) return id
+  }
+
+  for (const peer of opts.peers) {
+    if (bareIpv4(peer.address) === exporter) return peer.serverId
+    if (peer.allowedIps.some((cidr) => ipInCidr(exporter, cidr) || bareIpv4(cidr) === exporter)) {
+      return peer.serverId
+    }
+  }
+
+  const byHost = opts.hostIps.get(exporter)
+  if (byHost != null) return byHost
+
+  if (!isNatMasqueradeExporter(exporter, opts.overlayPrefix)) return null
+
+  const tunnelIds = [...new Set(opts.byTunnelIp.values())]
+  if (tunnelIds.length === 1) return tunnelIds[0] ?? null
+  const peerIds = [...new Set(opts.peers.map((p) => p.serverId))]
+  if (peerIds.length === 1) return peerIds[0] ?? null
+  return null
+}

backend/src/services/traffic-flow-overlay.ts

@@ -14,10 +14,12 @@ import {
   toRosBody,
 } from "./wireguard-ros.js"
 import {
+  enableTrafficFlowIngest,
   ensureHostKeys,
   getTrafficFlowSettingsRow,
   upsertHostPeer,
 } from "./traffic-flow-settings.js"
+import { startTrafficFlowListener } from "./traffic-flow-ingest.js"
 import { listTrafficFlowHostFiles } from "./traffic-flow-host-files.js"
 
 const IFACE_NAME = "wg-flow"
@@ -92,7 +94,14 @@ async function ensureWgInputAccept(client: MikrotikClient, listenPort: number):
   return true
 }
 
-async function ensureTrafficFlow(client: MikrotikClient, collectorIp: string, port: number): Promise<void> {
+/** Официальный авто-source UDP IPFIX, не фильтр 0.0.0.0/0. */
+export const FLOW_TARGET_SRC_AUTO = "0.0.0.0"
+
+async function ensureTrafficFlow(
+  client: MikrotikClient,
+  collectorIp: string,
+  port: number,
+): Promise<void> {
   const body = toRosBody({
     enabled: "yes",
     interfaces: "all",
@@ -111,6 +120,7 @@ async function ensureTrafficFlow(client: MikrotikClient, collectorIp: string, po
   const existing = targets.find((t) => String(t["dst-address"] ?? "") === collectorIp)
   const targetBody = toRosBody({
     "dst-address": collectorIp,
+    "src-address": FLOW_TARGET_SRC_AUTO,
     port: String(port),
     version: "ipfix",
   })
@@ -233,7 +243,7 @@ export async function applyFlowOverlay(
     }
 
     await ensureTrafficFlow(client, settings.collectorIp, settings.flowListenPort)
-    steps.push(`Traffic Flow → ${settings.collectorIp}:${settings.flowListenPort} ipfix`)
+    steps.push(`Traffic Flow → ${settings.collectorIp}:${settings.flowListenPort} ipfix (src auto)`)
 
     const listed = await listWireGuardInterfaces({ serverId: String(server.id), includePrivateKey: false })
     const created = listed.interfaces.find((i) => i.name === IFACE_NAME)
@@ -256,6 +266,10 @@ export async function applyFlowOverlay(
       endpoint: peerEndpoint,
     })
 
+    enableTrafficFlowIngest()
+    startTrafficFlowListener()
+    steps.push("Коллектор IPFIX на MM включён")
+
     return {
       ok: true,
       serverId: server.id,

backend/src/services/traffic-flow-parse.test.ts

@@ -1,6 +1,6 @@
 import assert from "node:assert/strict"
 import { parseFlowPacket, protoName, resetFlowTemplatesForTests } from "./traffic-flow-parse.js"
-import { allocateOverlayAddress } from "./traffic-flow-overlay.js"
+import { allocateOverlayAddress, FLOW_TARGET_SRC_AUTO, usablePublicHost } from "./traffic-flow-overlay.js"
 
 function netflowV5One(): Buffer {
   const buf = Buffer.alloc(24 + 48)
@@ -24,6 +24,7 @@ assert.equal(flows[0]?.src, "10.1.1.8")
 assert.equal(flows[0]?.dst, "8.8.8.8")
 assert.equal(flows[0]?.proto, 6)
 assert.equal(flows[0]?.bytes, 1500)
+assert.equal(flows[0]?.inIface, "1")
 assert.equal(protoName(6), "TCP")
 assert.equal(parseFlowPacket(Buffer.from([0, 1]), "1.1.1.1").length, 0)
 
@@ -31,11 +32,72 @@ const taken = new Set(["10.255.254.2"])
 assert.equal(allocateOverlayAddress("10.255.254.0/24", "10.255.254.1", 1, taken), "10.255.254.3")
 assert.equal(allocateOverlayAddress("10.255.254.0/24", "10.255.254.1", 2, new Set()), "10.255.254.3")
 
-import { usablePublicHost } from "./traffic-flow-overlay.js"
 assert.equal(usablePublicHost("localhost:8000"), "")
 assert.equal(usablePublicHost("127.0.0.1"), "")
 assert.equal(usablePublicHost("192.168.1.10"), "")
 assert.equal(usablePublicHost("mm.example.com:443"), "mm.example.com")
 assert.equal(usablePublicHost("203.0.113.10"), "203.0.113.10")
+assert.equal(FLOW_TARGET_SRC_AUTO, "0.0.0.0")
+
+resetFlowTemplatesForTests()
+{
+  const tpl = Buffer.alloc(16 + 16 + 20)
+  tpl.writeUInt16BE(10, 0)
+  tpl.writeUInt16BE(tpl.length, 2)
+  tpl.writeUInt16BE(2, 16)
+  tpl.writeUInt16BE(16, 18)
+  tpl.writeUInt16BE(256, 20)
+  tpl.writeUInt16BE(2, 22)
+  tpl.writeUInt16BE(8, 24)
+  tpl.writeUInt16BE(4, 26)
+  tpl.writeUInt16BE(12, 28)
+  tpl.writeUInt16BE(4, 30)
+  const data = Buffer.alloc(16 + 12)
+  data.writeUInt16BE(10, 0)
+  data.writeUInt16BE(data.length, 2)
+  data.writeUInt16BE(256, 16)
+  data.writeUInt16BE(12, 18)
+  data[20] = 10; data[21] = 1; data[22] = 1; data[23] = 8
+  data[24] = 8; data[25] = 8; data[26] = 8; data[27] = 8
+  const fromTpl = parseFlowPacket(tpl, "172.18.0.2")
+  assert.equal(fromTpl.length, 0)
+  const fromData = parseFlowPacket(data, "172.18.0.2")
+  assert.equal(fromData.length, 1)
+  assert.equal(fromData[0]?.src, "10.1.1.8")
+  assert.equal(fromData[0]?.dst, "8.8.8.8")
+}
+
+resetFlowTemplatesForTests()
+{
+  const tpl = Buffer.alloc(16 + 24)
+  tpl.writeUInt16BE(10, 0)
+  tpl.writeUInt16BE(tpl.length, 2)
+  tpl.writeUInt16BE(2, 16)
+  tpl.writeUInt16BE(24, 18)
+  tpl.writeUInt16BE(256, 20)
+  tpl.writeUInt16BE(4, 22)
+  tpl.writeUInt16BE(8, 24)
+  tpl.writeUInt16BE(4, 26)
+  tpl.writeUInt16BE(12, 28)
+  tpl.writeUInt16BE(4, 30)
+  tpl.writeUInt16BE(10, 32)
+  tpl.writeUInt16BE(4, 34)
+  tpl.writeUInt16BE(82, 36)
+  tpl.writeUInt16BE(6, 38)
+  const data = Buffer.alloc(16 + 22)
+  data.writeUInt16BE(10, 0)
+  data.writeUInt16BE(data.length, 2)
+  data.writeUInt16BE(256, 16)
+  data.writeUInt16BE(22, 18)
+  data[20] = 10; data[21] = 1; data[22] = 1; data[23] = 8
+  data[24] = 8; data[25] = 8; data[26] = 8; data[27] = 8
+  data.writeUInt32BE(13, 28)
+  data.write("ether1", 32)
+  parseFlowPacket(tpl, "10.255.254.3")
+  const named = parseFlowPacket(data, "10.255.254.3")
+  assert.equal(named.length, 1)
+  assert.equal(named[0]?.inIface, "13")
+  assert.equal(named[0]?.src, "10.1.1.8")
+}
 
 console.log("traffic-flow-parse.test.ts: ok")

backend/src/services/traffic-flow-parse.ts

@@ -7,6 +7,7 @@ export interface ParsedFlow {
   bytes: number
   packets: number
   inIface: string
+  outIface: string
 }
 
 interface FieldSpec {
@@ -24,6 +25,48 @@ function ipv4(buf: Buffer, offset: number): string {
   return `${buf[offset]}.${buf[offset + 1]}.${buf[offset + 2]}.${buf[offset + 3]}`
 }
 
+function ipv6(buf: Buffer, offset: number): string {
+  const parts: string[] = []
+  for (let i = 0; i < 8; i++) parts.push(buf.readUInt16BE(offset + i * 2).toString(16))
+  return parts.join(":")
+}
+
+const VAR_LEN = 0xffff
+
+function consumeField(
+  buf: Buffer,
+  off: number,
+  length: number,
+  limit: number,
+): { data: Buffer; next: number } | null {
+  if (length === VAR_LEN) {
+    if (off >= limit) return null
+    const first = buf[off]!
+    if (first < 255) {
+      const end = off + 1 + first
+      if (end > limit) return null
+      return { data: buf.subarray(off + 1, end), next: end }
+    }
+    if (off + 3 > limit) return null
+    const len = buf.readUInt16BE(off + 1)
+    const end = off + 3 + len
+    if (end > limit) return null
+    return { data: buf.subarray(off + 3, end), next: end }
+  }
+  const end = off + length
+  if (end > limit) return null
+  return { data: buf.subarray(off, end), next: end }
+}
+
+function fixedRecordSize(fields: FieldSpec[]): number | null {
+  let n = 0
+  for (const f of fields) {
+    if (f.length === VAR_LEN) return null
+    n += f.length
+  }
+  return n
+}
+
 function readUint(buf: Buffer, offset: number, length: number): number {
   if (length === 1) return buf.readUInt8(offset)
   if (length === 2) return buf.readUInt16BE(offset)
@@ -53,6 +96,7 @@ function parseNetflowV5(buf: Buffer): ParsedFlow[] {
       dstPort: buf.readUInt16BE(off + 34),
       proto: buf.readUInt8(off + 38),
       inIface: String(buf.readUInt16BE(off + 12)),
+      outIface: String(buf.readUInt16BE(off + 14)),
     })
     off += 48
   }
@@ -87,7 +131,12 @@ function parseIpfixTemplates(exporter: string, buf: Buffer, setStart: number, se
   templatesByExporter.set(exporter, map)
 }
 
-function recordFromFields(fields: FieldSpec[], buf: Buffer, offset: number): { flow: ParsedFlow; next: number } | null {
+function recordFromFields(
+  fields: FieldSpec[],
+  buf: Buffer,
+  offset: number,
+  limit: number,
+): { flow: ParsedFlow; next: number } | null {
   let off = offset
   let src = ""
   let dst = ""
@@ -97,40 +146,86 @@ function recordFromFields(fields: FieldSpec[], buf: Buffer, offset: number): { f
   let bytes = 0
   let packets = 0
   let inIface = ""
+  let outIface = ""
+  let ifaceName = ""
   for (const f of fields) {
-    if (off + f.length > buf.length) return null
+    const field = consumeField(buf, off, f.length, limit)
+    if (!field) return null
+    const { data } = field
     switch (f.type) {
       case 8:
-        if (f.length === 4) src = ipv4(buf, off)
+        if (data.length === 4) src = ipv4(data, 0)
         break
       case 12:
-        if (f.length === 4) dst = ipv4(buf, off)
+        if (data.length === 4) dst = ipv4(data, 0)
+        break
+      case 27:
+        if (data.length === 16 && !src) src = ipv6(data, 0)
+        break
+      case 28:
+        if (data.length === 16 && !dst) dst = ipv6(data, 0)
+        break
+      case 225:
+        if (data.length === 4 && !src) src = ipv4(data, 0)
+        break
+      case 226:
+        if (data.length === 4 && !dst) dst = ipv4(data, 0)
         break
       case 4:
-        proto = readUint(buf, off, f.length)
+        proto = readUint(data, 0, data.length)
         break
       case 7:
-        srcPort = readUint(buf, off, f.length)
+        srcPort = readUint(data, 0, data.length)
         break
       case 11:
-        dstPort = readUint(buf, off, f.length)
+        dstPort = readUint(data, 0, data.length)
         break
       case 1:
-        bytes = readUint(buf, off, f.length)
+        bytes = readUint(data, 0, data.length)
         break
       case 2:
-        packets = readUint(buf, off, f.length)
+        packets = readUint(data, 0, data.length)
+        break
+      case 85:
+        if (!bytes) bytes = readUint(data, 0, data.length)
+        break
+      case 86:
+        if (!packets) packets = readUint(data, 0, data.length)
         break
       case 10:
-        inIface = String(readUint(buf, off, f.length))
+        inIface = String(readUint(data, 0, data.length))
+        break
+      case 14:
+        outIface = String(readUint(data, 0, data.length))
+        break
+      case 82:
+        ifaceName = data.toString("utf8").replace(/\0/g, "").trim()
         break
       default:
         break
     }
-    off += f.length
+    off = field.next
+  }
+  if (!inIface && ifaceName) inIface = ifaceName
+  return { flow: { src, dst, proto, srcPort, dstPort, bytes, packets, inIface, outIface }, next: off }
+}
+
+function parseDataRecords(
+  tpl: Template,
+  buf: Buffer,
+  recOff: number,
+  setEnd: number,
+  out: ParsedFlow[],
+) {
+  const size = fixedRecordSize(tpl.fields)
+  while (recOff + 1 < setEnd) {
+    if (size != null && recOff + size > setEnd) break
+    const parsed = recordFromFields(tpl.fields, buf, recOff, setEnd)
+    if (!parsed) break
+    if (parsed.flow.src || parsed.flow.dst) out.push(parsed.flow)
+    if (parsed.next <= recOff) break
+    recOff = parsed.next
   }
-  if (!src && !dst) return { flow: { src, dst, proto, srcPort, dstPort, bytes, packets, inIface }, next: off }
-  return { flow: { src, dst, proto, srcPort, dstPort, bytes, packets, inIface }, next: off }
 }
 
 function parseIpfix(buf: Buffer, exporter: string): ParsedFlow[] {
@@ -148,16 +243,7 @@ function parseIpfix(buf: Buffer, exporter: string): ParsedFlow[] {
       parseIpfixTemplates(exporter, buf, off, setEnd, setId)
     } else if (setId >= 256) {
       const tpl = templatesByExporter.get(exporter)?.get(setId)
-      if (tpl) {
-        let recOff = off + 4
-        while (recOff + 1 < setEnd) {
-          const parsed = recordFromFields(tpl.fields, buf, recOff)
-          if (!parsed) break
-          if (parsed.flow.src || parsed.flow.dst) out.push(parsed.flow)
-          if (parsed.next <= recOff) break
-          recOff = parsed.next
-        }
-      }
+      if (tpl) parseDataRecords(tpl, buf, off + 4, setEnd, out)
     }
     off = setEnd
   }
@@ -191,16 +277,7 @@ function parseNetflowV9(buf: Buffer, exporter: string): ParsedFlow[] {
       templatesByExporter.set(exporter, map)
     } else if (setId >= 256) {
       const tpl = map.get(setId)
-      if (tpl) {
-        let recOff = off + 4
-        while (recOff + 1 < setEnd) {
-          const parsed = recordFromFields(tpl.fields, buf, recOff)
-          if (!parsed) break
-          if (parsed.flow.src || parsed.flow.dst) out.push(parsed.flow)
-          if (parsed.next <= recOff) break
-          recOff = parsed.next
-        }
-      }
+      if (tpl) parseDataRecords(tpl, buf, off + 4, setEnd, out)
     }
     off = setEnd
   }

backend/src/services/traffic-flow-settings.ts

@@ -110,7 +110,6 @@ export function recordFlowPacket(exporterIp: string) {
     lastDatagramAt: nowIso(),
     lastExporterIp: exporterIp,
     packetsReceived: row.packetsReceived + 1,
-    lastError: "",
     updatedAt: nowIso(),
   }).where(eq(trafficFlowSettings.id, 1)).run()
 }
@@ -122,6 +121,13 @@ export function recordFlowListenerError(message: string) {
   }).where(eq(trafficFlowSettings.id, 1)).run()
 }
 
+export function enableTrafficFlowIngest() {
+  db.update(trafficFlowSettings).set({
+    enabled: true,
+    updatedAt: nowIso(),
+  }).where(eq(trafficFlowSettings.id, 1)).run()
+}
+
 export function listHostPeers(): FlowHostPeer[] {
   return parsePeers(getTrafficFlowSettingsRow().peersJson)
 }

components/data-grids/traffic-flows-data-grid.tsx

@@ -19,7 +19,13 @@ function formatBytes(n: number): string {
   return `${n} Б`
 }
 
-function TrafficFlowsDataGrid({ rows }: { rows: FlowTalkerDto[] }) {
+function TrafficFlowsDataGrid({
+  rows,
+  emptyHint,
+}: {
+  rows: FlowTalkerDto[]
+  emptyHint?: string
+}) {
   const columns = useMemo<ColumnDef<FlowTalkerDto>[]>(
     () => [
       {
@@ -53,6 +59,15 @@ function TrafficFlowsDataGrid({ rows }: { rows: FlowTalkerDto[] }) {
         ),
         meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
       },
+      {
+        id: "app",
+        accessorFn: (r) => r.application ?? r.protoName,
+        header: () => <span className="text-xs font-medium text-muted-foreground">App</span>,
+        cell: ({ row }) => (
+          <span className="text-xs">{row.original.application ?? row.original.protoName}</span>
+        ),
+        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
+      },
       {
         id: "proto",
         accessorKey: "protoName",
@@ -96,7 +111,10 @@ function TrafficFlowsDataGrid({ rows }: { rows: FlowTalkerDto[] }) {
     <DataGridShell
       table={table}
       recordCount={rows.length}
-      emptyMessage="Пока нет IPFIX. Поднимите wg-flow на хосте MM и подключите jump-host одним кликом."
+      emptyMessage={
+        emptyHint
+        || "Пока нет IPFIX. Поднимите wg-flow на хосте MM и подключите jump-host одним кликом."
+      }
     />
   )
 }

components/traffic/flow-analytics-panel.tsx

@@ -0,0 +1,369 @@
+"use client"
+
+import { useMemo, useState } from "react"
+import { type ColumnDef, getCoreRowModel, useReactTable } from "@tanstack/react-table"
+import type { FlowAnalyticsDto, FlowBreakdownRow, FlowEntityCard } from "@mmapp/contracts/traffic-flow"
+import { ArrowDownIcon, ArrowUpIcon, GitBranchIcon } from "lucide-react"
+import { Badge } from "@/components/reui/badge"
+import { KpiStatGrid } from "@/components/reui-kit/kpi-stat-grid"
+import { TrafficRxTxChart } from "@/components/reui-kit/traffic-rx-tx-chart"
+import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"
+import { DataGridShell } from "@/components/data-grids/shared/data-grid-shell"
+import {
+  DATA_GRID_CELL_PAD,
+  DATA_GRID_CELL_PAD_FIRST,
+  DATA_GRID_CELL_PAD_LAST,
+} from "@/components/data-grids/shared/data-grid-layout"
+import { TrafficFlowsDataGrid } from "@/components/data-grids/traffic-flows-data-grid"
+import { Sparkline } from "@/components/sparkline"
+import { StatusDot } from "@/components/status-dot"
+import { Flag } from "@/components/flag"
+import { fmtRate } from "@/lib/fmt-rate"
+import { cn } from "@/lib/utils"
+
+function formatBytes(n: number): string {
+  if (n >= 1_000_000_000) return `${(n / 1_000_000_000).toFixed(2)} ГБ`
+  if (n >= 1_000_000) return `${(n / 1_000_000).toFixed(1)} МБ`
+  if (n >= 1000) return `${(n / 1000).toFixed(1)} КБ`
+  return `${n} Б`
+}
+
+const RANGE_KEYS = ["5m", "15m", "1h", "4h", "24h"] as const
+const RANGE_LABELS: Record<string, string> = {
+  "5m": "5м",
+  "15m": "15м",
+  "1h": "1ч",
+  "4h": "4ч",
+  "24h": "24ч",
+}
+
+function MiniAreaChart({ rx, tx, height = 44 }: { rx: number[]; tx: number[]; height?: number }) {
+  const W = 300
+  const H = height
+  const maxVal = Math.max(...rx, ...tx, 1) * 1.1
+  const xAt = (i: number) => (rx.length <= 1 ? 0 : (i / (rx.length - 1)) * W)
+  const yAt = (v: number) => H - (v / maxVal) * H
+  const area = (arr: number[]) => {
+    const pts = arr.map((v, i) => `${xAt(i).toFixed(1)},${yAt(v).toFixed(1)}`).join(" L ")
+    return `M 0,${H} L ${pts} L ${W},${H} Z`
+  }
+  const line = (arr: number[]) => arr.map((v, i) => `${xAt(i).toFixed(1)},${yAt(v).toFixed(1)}`).join(" ")
+  return (
+    <svg viewBox={`0 0 ${W} ${H}`} className="w-full h-11" preserveAspectRatio="none">
+      <path d={area(rx)} fill="var(--chart-rx)" fillOpacity="0.15" />
+      <polyline points={line(rx)} fill="none" stroke="var(--chart-rx)" strokeWidth="1.5" />
+      {tx.some((v) => v > 0) ? (
+        <polyline points={line(tx)} fill="none" stroke="var(--chart-tx)" strokeWidth="1.5" />
+      ) : null}
+    </svg>
+  )
+}
+
+export function FlowEntityCardView({
+  card,
+  selected,
+  onClick,
+}: {
+  card: FlowEntityCard
+  selected: boolean
+  onClick: () => void
+}) {
+  return (
+    <button
+      type="button"
+      onClick={onClick}
+      className={cn(
+        "text-left w-full rounded-lg border p-3 transition-colors hover:bg-muted/50",
+        selected ? "border-primary bg-primary/5" : "border-border bg-card",
+        card.status === "offline" && "opacity-60",
+      )}
+    >
+      <div className="flex items-center justify-between gap-2 mb-2">
+        <div className="flex items-center gap-1.5 min-w-0">
+          <StatusDot status={card.status} />
+          <span className="text-xs font-medium truncate">{card.name}</span>
+        </div>
+        <span className="text-[10px] font-mono text-muted-foreground shrink-0 flex items-center gap-1">
+          {card.country !== "UN" ? <Flag code={card.country} /> : null}
+          {card.site}
+        </span>
+      </div>
+      <MiniAreaChart rx={card.rxSeries} tx={card.txSeries} />
+      <div className="flex justify-between mt-2 gap-2">
+        <div className="flex items-center gap-1 text-[11px]">
+          <ArrowDownIcon className="size-3 text-success" />
+          <span className="font-mono font-medium text-success">{fmtRate(card.rxNow)}</span>
+        </div>
+        <div className="flex items-center gap-1 text-[11px]">
+          <ArrowUpIcon className="size-3 text-info" />
+          <span className="font-mono font-medium text-info">{fmtRate(card.txNow)}</span>
+        </div>
+        <div className="flex items-center gap-0.5 text-[10px] text-muted-foreground">
+          <GitBranchIcon className="size-3" />{card.sessions}
+        </div>
+      </div>
+    </button>
+  )
+}
+
+function FlowBreakdownGrid({ rows, empty }: { rows: FlowBreakdownRow[]; empty?: string }) {
+  const columns = useMemo<ColumnDef<FlowBreakdownRow>[]>(
+    () => [
+      {
+        id: "label",
+        accessorKey: "label",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Имя</span>,
+        cell: ({ row }) => (
+          <div className="flex min-w-0 flex-col gap-1">
+            <span className="text-sm font-medium truncate">{row.original.label}</span>
+            <div className="h-1 rounded-full bg-muted overflow-hidden">
+              <div
+                className="h-full bg-primary"
+                style={{ width: `${Math.min(100, row.original.percent)}%` }}
+              />
+            </div>
+          </div>
+        ),
+        meta: { headerClassName: DATA_GRID_CELL_PAD_FIRST, cellClassName: DATA_GRID_CELL_PAD_FIRST },
+      },
+      {
+        id: "share",
+        accessorKey: "percent",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Доля</span>,
+        cell: ({ row }) => (
+          <span className="text-xs tabular-nums">{row.original.percent.toFixed(1)}%</span>
+        ),
+        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
+      },
+      {
+        id: "rate",
+        accessorFn: (r) => r.bps,
+        header: () => <span className="text-xs font-medium text-muted-foreground">Скорость</span>,
+        cell: ({ row }) => (
+          <span className="text-xs tabular-nums">{fmtRate(row.original.bps / 1_000_000)}</span>
+        ),
+        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
+      },
+      {
+        id: "bytes",
+        accessorKey: "bytes",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Байты</span>,
+        cell: ({ row }) => (
+          <span className="text-xs tabular-nums">{formatBytes(row.original.bytes)}</span>
+        ),
+        meta: { headerClassName: DATA_GRID_CELL_PAD_LAST, cellClassName: DATA_GRID_CELL_PAD_LAST },
+      },
+    ],
+    [],
+  )
+
+  const table = useReactTable({
+    data: rows,
+    columns,
+    getCoreRowModel: getCoreRowModel(),
+    getRowId: (row) => row.id,
+  })
+
+  return (
+    <DataGridShell table={table} recordCount={rows.length} emptyMessage={empty ?? "Нет данных за период"} />
+  )
+}
+
+export function FlowAnalyticsDetail({
+  card,
+  analytics,
+  range,
+  onRange,
+  selectedIface,
+  onIface,
+  liveHint,
+  emptyHint,
+}: {
+  card: FlowEntityCard | null
+  analytics: FlowAnalyticsDto | null
+  range: string
+  onRange: (r: string) => void
+  selectedIface: string
+  onIface: (name: string) => void
+  liveHint?: string
+  emptyHint?: string
+}) {
+  const [slice, setSlice] = useState("applications")
+  const rxNow = analytics ? analytics.bpsNow / 1_000_000 : (card?.rxNow ?? 0)
+  const bytes = analytics?.bytes ?? card?.bytes ?? 0
+
+  if (!card) {
+    return (
+      <p className="text-sm text-muted-foreground py-8 text-center">
+        Выберите сервер или клиента слева
+      </p>
+    )
+  }
+
+  return (
+    <>
+      <div className="flex items-start justify-between mb-3 gap-3">
+        <div className="flex items-center gap-2 flex-wrap min-w-0">
+          <StatusDot status={card.status} />
+          <h2 className="text-base font-semibold truncate">{card.name}</h2>
+          <span className="text-xs font-mono bg-muted px-1.5 py-0.5 rounded flex items-center gap-1">
+            {card.country !== "UN" ? <Flag code={card.country} /> : null}
+            {card.site}
+          </span>
+        </div>
+        <div className="flex gap-1 shrink-0">
+          {RANGE_KEYS.map((r) => (
+            <button
+              key={r}
+              type="button"
+              onClick={() => onRange(r)}
+              className={cn(
+                "h-7 px-2 text-xs rounded border transition-colors",
+                range === r
+                  ? "border-primary bg-primary/10 text-primary font-medium"
+                  : "border-border text-muted-foreground hover:text-foreground",
+              )}
+            >
+              {RANGE_LABELS[r]}
+            </button>
+          ))}
+        </div>
+      </div>
+
+      {analytics?.ifaces && analytics.ifaces.length > 0 ? (
+        <div className="mb-3 pb-3 border-b">
+          <div className="flex items-center gap-2 flex-wrap">
+            <span className="text-[11px] text-muted-foreground mr-1">Интерфейс:</span>
+            <button
+              type="button"
+              onClick={() => onIface("__all__")}
+              className={cn(
+                "inline-flex items-center gap-1 rounded-full border px-2.5 py-0.5 text-[10px] font-medium transition-all",
+                selectedIface === "__all__"
+                  ? "bg-foreground text-background border-foreground"
+                  : "border-border text-muted-foreground hover:text-foreground hover:border-foreground/40",
+              )}
+            >
+              Все
+            </button>
+            {analytics.ifaces.map((iface) => {
+              const active = selectedIface === iface.name
+              return (
+                <button
+                  key={`${iface.name}:${iface.index}`}
+                  type="button"
+                  onClick={() => onIface(iface.name)}
+                  className={cn(
+                    "inline-flex items-center gap-1.5 rounded-full border px-2.5 py-0.5 text-[10px] font-medium transition-all",
+                    active
+                      ? "bg-foreground text-background border-foreground"
+                      : "border-border text-muted-foreground hover:text-foreground hover:border-foreground/40",
+                  )}
+                >
+                  {iface.name}
+                </button>
+              )
+            })}
+          </div>
+        </div>
+      ) : null}
+
+      <TrafficRxTxChart
+        rx={analytics?.rxSeries ?? card.rxSeries}
+        tx={analytics?.txSeries ?? card.txSeries}
+        range={range}
+      />
+
+      <div className="mt-4 pt-4 border-t">
+        <KpiStatGrid
+          aria-label="Скорость потоков"
+          items={[
+            {
+              id: "bps-now",
+              label: "Скорость сейчас",
+              value: fmtRate(rxNow),
+              hint: liveHint,
+              icon: <ArrowDownIcon className="size-4" />,
+              iconClassName: "text-success",
+            },
+            {
+              id: "bytes",
+              label: "Байт за период",
+              value: formatBytes(bytes),
+              icon: <ArrowUpIcon className="size-4" />,
+              iconClassName: "text-info",
+            },
+            {
+              id: "flows",
+              label: "Разговоры",
+              value: String(analytics?.conversations ?? card.sessions),
+              icon: <GitBranchIcon className="size-4" />,
+              iconClassName: "text-warning",
+            },
+          ]}
+        />
+      </div>
+
+      <div className="mt-4 pt-4 border-t flex flex-col gap-3">
+        <div className="flex items-center justify-between gap-2">
+          <p className="text-xs text-muted-foreground">Аналитика потребления</p>
+          {analytics?.live ? <Badge variant="success-light" size="sm">live</Badge> : null}
+        </div>
+        <Tabs value={slice} onValueChange={(v) => setSlice(String(v))} className="gap-3">
+          <TabsList variant="line" className="flex flex-wrap h-auto">
+            <TabsTrigger value="applications">Приложения</TabsTrigger>
+            <TabsTrigger value="protocols">Протоколы</TabsTrigger>
+            <TabsTrigger value="sources">Источники</TabsTrigger>
+            <TabsTrigger value="destinations">Назначения</TabsTrigger>
+            <TabsTrigger value="conversations">Разговоры</TabsTrigger>
+            <TabsTrigger value="interfaces">Интерфейсы</TabsTrigger>
+          </TabsList>
+          <TabsContent value="applications">
+            <FlowBreakdownGrid rows={analytics?.applications ?? []} />
+          </TabsContent>
+          <TabsContent value="protocols">
+            <FlowBreakdownGrid rows={analytics?.protocols ?? []} />
+          </TabsContent>
+          <TabsContent value="sources">
+            <FlowBreakdownGrid rows={analytics?.sources ?? []} />
+          </TabsContent>
+          <TabsContent value="destinations">
+            <FlowBreakdownGrid rows={analytics?.destinations ?? []} />
+          </TabsContent>
+          <TabsContent value="conversations">
+            <TrafficFlowsDataGrid
+              rows={analytics?.conversationsList ?? []}
+              emptyHint={emptyHint}
+            />
+          </TabsContent>
+          <TabsContent value="interfaces">
+            <FlowBreakdownGrid rows={analytics?.interfaces ?? []} />
+          </TabsContent>
+        </Tabs>
+      </div>
+
+      <div className="mt-4 pt-4 border-t grid grid-cols-2 gap-4">
+        <div>
+          <p className="text-xs text-muted-foreground mb-1">Объём за период</p>
+          <p className="text-lg font-semibold tabular-nums">{formatBytes(bytes)}</p>
+          <Sparkline
+            data={analytics?.rxSeries ?? card.rxSeries}
+            width={180}
+            height={28}
+            color="var(--chart-rx)"
+            filled
+          />
+        </div>
+        <div>
+          <p className="text-xs text-muted-foreground mb-1">Уник. адреса</p>
+          <p className="text-lg font-semibold tabular-nums">
+            {analytics?.uniqueSrc ?? 0}
+            <span className="text-sm font-normal text-muted-foreground"> src · </span>
+            {analytics?.uniqueDst ?? 0}
+            <span className="text-sm font-normal text-muted-foreground"> dst</span>
+          </p>
+        </div>
+      </div>
+    </>
+  )
+}

deploy/docker-compose.cdn-mm.yml

@@ -136,8 +136,9 @@ services:
       AUTH_PORTAL_URL: ${AUTH_PORTAL_URL:-https://auth.shnt.top}
       # IPFIX: внутри контейнера слушать все iface; на хосте bind только WG-IP после wg-quick@wg-flow
       FLOW_LISTEN_HOST: "0.0.0.0"
-    # ports:
-    #   - "10.255.254.1:4739:4739/udp"
+    # Сначала wg-quick@wg-flow (адрес 10.255.254.1), затем recreate backend.
+    ports:
+      - "10.255.254.1:4739:4739/udp"
     volumes:
       - ./data/mm:/app/data
     networks:

hooks/use-flow-live.ts

@@ -0,0 +1,90 @@
+"use client"
+
+import { useEffect, useState } from "react"
+import type { FlowAnalyticsDto } from "@mmapp/contracts/traffic-flow"
+import { resolveApiUrl, withAuthHeaders } from "@/shared/api/http-client"
+import { flowQuery } from "@/shared/api/traffic-flow"
+
+function parseSseBlock(block: string): { event: string; data: string } {
+  let event = "message"
+  const dataLines: string[] = []
+  for (const line of block.split("\n")) {
+    if (line.startsWith("event:")) event = line.slice(6).trim()
+    else if (line.startsWith("data:")) dataLines.push(line.slice(5).trim())
+  }
+  return { event, data: dataLines.join("\n") }
+}
+
+export function useFlowLive(opts: {
+  enabled: boolean
+  backendUrl: string
+  range: string
+  serverId?: string
+  userId?: string
+  iface?: string
+}): { sample: FlowAnalyticsDto | null; error: string | null } {
+  const [sample, setSample] = useState<FlowAnalyticsDto | null>(null)
+  const [error, setError] = useState<string | null>(null)
+
+  useEffect(() => {
+    if (!opts.enabled) {
+      setSample(null)
+      setError(null)
+      return
+    }
+
+    const ac = new AbortController()
+    setSample(null)
+    setError(null)
+    const path = `/api/traffic/flow/live${flowQuery({
+      range: opts.range,
+      serverId: opts.serverId,
+      userId: opts.userId,
+      iface: opts.iface,
+    })}`
+    const url = resolveApiUrl(opts.backendUrl, path)
+
+    let buf = ""
+
+    void (async () => {
+      try {
+        const res = await fetch(url, {
+          headers: withAuthHeaders({ Accept: "text/event-stream" }),
+          signal: ac.signal,
+          credentials: "include",
+        })
+        if (!res.ok || !res.body) {
+          setError(`live HTTP ${res.status}`)
+          return
+        }
+        const reader = res.body.getReader()
+        const decoder = new TextDecoder()
+        while (!ac.signal.aborted) {
+          const { done, value } = await reader.read()
+          if (done) break
+          buf += decoder.decode(value, { stream: true })
+          const parts = buf.split("\n\n")
+          buf = parts.pop() ?? ""
+          for (const raw of parts) {
+            if (!raw.trim() || raw.trim().startsWith(":")) continue
+            const ev = parseSseBlock(raw)
+            if (ev.event === "sample" && ev.data) {
+              setSample(JSON.parse(ev.data) as FlowAnalyticsDto)
+              setError(null)
+            } else if (ev.event === "error" && ev.data) {
+              const parsed = JSON.parse(ev.data) as { error?: string }
+              setError(parsed.error ?? "live error")
+            }
+          }
+        }
+      } catch (e) {
+        if (ac.signal.aborted) return
+        setError(e instanceof Error ? e.message : "live error")
+      }
+    })()
+
+    return () => ac.abort()
+  }, [opts.enabled, opts.backendUrl, opts.range, opts.serverId, opts.userId, opts.iface])
+
+  return { sample, error }
+}

packages/contracts/src/traffic-flow.ts

@@ -79,6 +79,8 @@ export const flowTalkerDtoSchema = z.object({
   packets: z.number().nonnegative(),
   bps: z.number().nonnegative(),
   inIface: z.string(),
+  inIfaceIndex: z.string().optional(),
+  application: z.string().optional(),
 })
 
 export const flowStatsDtoSchema = z.object({
@@ -88,6 +90,12 @@ export const flowStatsDtoSchema = z.object({
   uniqueDst: z.number().int().nonnegative(),
   topProto: z.string(),
   talkers: z.array(flowTalkerDtoSchema),
+  lastExporterIp: z.string().nullable().optional(),
+  lastError: z.string().nullable().optional(),
+  packetsReceived: z.number().int().nonnegative().optional(),
+  lastDatagramAt: z.string().nullable().optional(),
+  listenerBound: z.boolean().optional(),
+  listenerAddress: z.string().nullable().optional(),
 })
 
 export type FlowHostPeer = z.infer<typeof flowHostPeerSchema>
@@ -95,5 +103,75 @@ export type TrafficFlowSettingsDto = z.infer<typeof trafficFlowSettingsDtoSchema
 export type TrafficFlowSettingsPatch = z.infer<typeof trafficFlowSettingsPatchSchema>
 export type TrafficFlowOverlayResult = z.infer<typeof trafficFlowOverlayResultSchema>
 export type TrafficFlowHostFile = z.infer<typeof trafficFlowHostFileSchema>
+export const flowBreakdownRowSchema = z.object({
+  id: z.string(),
+  label: z.string(),
+  bytes: z.number().nonnegative(),
+  packets: z.number().nonnegative(),
+  bps: z.number().nonnegative(),
+  percent: z.number().nonnegative(),
+})
+
+export const flowIfaceChipSchema = z.object({
+  name: z.string(),
+  index: z.string(),
+  bps: z.number().nonnegative(),
+})
+
+export const flowEntityCardSchema = z.object({
+  id: z.string(),
+  name: z.string(),
+  subtitle: z.string(),
+  site: z.string(),
+  country: z.string(),
+  status: z.enum(["online", "offline", "degraded"]),
+  rxNow: z.number(),
+  txNow: z.number(),
+  sessions: z.number().int().nonnegative(),
+  rxSeries: z.array(z.number()),
+  txSeries: z.array(z.number()),
+  bytes: z.number().nonnegative(),
+})
+
+export const flowAnalyticsDtoSchema = z.object({
+  bpsNow: z.number().nonnegative(),
+  bytes: z.number().nonnegative(),
+  packets: z.number().nonnegative(),
+  conversations: z.number().int().nonnegative(),
+  uniqueSrc: z.number().int().nonnegative(),
+  uniqueDst: z.number().int().nonnegative(),
+  topProto: z.string(),
+  rxSeries: z.array(z.number()),
+  txSeries: z.array(z.number()),
+  applications: z.array(flowBreakdownRowSchema),
+  protocols: z.array(flowBreakdownRowSchema),
+  sources: z.array(flowBreakdownRowSchema),
+  destinations: z.array(flowBreakdownRowSchema),
+  interfaces: z.array(flowBreakdownRowSchema),
+  conversationsList: z.array(flowTalkerDtoSchema),
+  ifaces: z.array(flowIfaceChipSchema),
+  live: z.boolean(),
+})
+
+export const flowExportersDtoSchema = z.object({
+  exporters: z.array(flowEntityCardSchema),
+  lastExporterIp: z.string().nullable().optional(),
+  lastError: z.string().nullable().optional(),
+  packetsReceived: z.number().int().nonnegative().optional(),
+  lastDatagramAt: z.string().nullable().optional(),
+  listenerBound: z.boolean().optional(),
+  listenerAddress: z.string().nullable().optional(),
+})
+
+export const flowClientsDtoSchema = z.object({
+  clients: z.array(flowEntityCardSchema),
+})
+
 export type FlowTalkerDto = z.infer<typeof flowTalkerDtoSchema>
 export type FlowStatsDto = z.infer<typeof flowStatsDtoSchema>
+export type FlowBreakdownRow = z.infer<typeof flowBreakdownRowSchema>
+export type FlowIfaceChip = z.infer<typeof flowIfaceChipSchema>
+export type FlowEntityCard = z.infer<typeof flowEntityCardSchema>
+export type FlowAnalyticsDto = z.infer<typeof flowAnalyticsDtoSchema>
+export type FlowExportersDto = z.infer<typeof flowExportersDtoSchema>
+export type FlowClientsDto = z.infer<typeof flowClientsDtoSchema>

shared/api/traffic-flow.ts

@@ -1,4 +1,7 @@
 import type {
+  FlowAnalyticsDto,
+  FlowClientsDto,
+  FlowExportersDto,
   FlowStatsDto,
   TrafficFlowHostFile,
   TrafficFlowOverlayResult,
@@ -50,3 +53,35 @@ export async function applyTrafficFlowOverlay(
 export async function getTrafficFlows(baseUrl: string, range = "5m"): Promise<FlowStatsDto> {
   return requestJson<FlowStatsDto>(baseUrl, `/api/traffic/flows?range=${encodeURIComponent(range)}`)
 }
+
+function flowQuery(params: {
+  range?: string
+  serverId?: string
+  userId?: string
+  iface?: string
+}): string {
+  const q = new URLSearchParams()
+  if (params.range) q.set("range", params.range)
+  if (params.serverId) q.set("serverId", params.serverId)
+  if (params.userId) q.set("userId", params.userId)
+  if (params.iface && params.iface !== "__all__") q.set("iface", params.iface)
+  const s = q.toString()
+  return s ? `?${s}` : ""
+}
+
+export async function getFlowExporters(baseUrl: string, range = "5m"): Promise<FlowExportersDto> {
+  return requestJson<FlowExportersDto>(baseUrl, `/api/traffic/flow/exporters?range=${encodeURIComponent(range)}`)
+}
+
+export async function getFlowClients(baseUrl: string, range = "5m"): Promise<FlowClientsDto> {
+  return requestJson<FlowClientsDto>(baseUrl, `/api/traffic/flow/clients?range=${encodeURIComponent(range)}`)
+}
+
+export async function getFlowAnalytics(
+  baseUrl: string,
+  params: { range?: string; serverId?: string; userId?: string; iface?: string },
+): Promise<FlowAnalyticsDto> {
+  return requestJson<FlowAnalyticsDto>(baseUrl, `/api/traffic/flow/analytics${flowQuery(params)}`)
+}
+
+export { flowQuery }