feat(traffic): добавить карту и срезы IPFIX по странам и сессиям

Имена ifIndex пишутся по обоим ключам REST, дедуп 5-tuple по max байт, RIPEstat только из prefix-кэша, карта откуда-куда в Frame.

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

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

@@ -5,7 +5,19 @@ import {
   resetFlowRingsForTests,
 } from "./traffic-flow-ingest.js"
 import { buildFlowAnalytics } from "./traffic-flow-analytics.js"
+import { disableCatalogFetchForTests, resetFlowCatalogForTests, seedFlowCatalogForTests } from "./traffic-flow-classify.js"
+import {
+  disableRipeEnqueueForTests,
+  disableRipePersistForTests,
+  resetRipeCacheForTests,
+  seedRipeCacheForTests,
+} from "./traffic-flow-ripe.js"
 
+disableCatalogFetchForTests()
+resetFlowCatalogForTests()
+disableRipePersistForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
 resetIfaceCacheForTests()
 resetFlowRingsForTests()
 rememberServerIfaces(7, [
@@ -65,4 +77,92 @@ try {
   resetIfaceCacheForTests()
 }
 
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+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: "8.8.8.8",
+    proto: 6,
+    srcPort: 51234,
+    dstPort: 443,
+    bytes: 9_000,
+    packets: 9,
+    inIface: "10",
+    outIface: "",
+  },
+])
+try {
+  const summed = buildFlowAnalytics({ minutes: 5, serverId: 7, dedup: false })
+  assert.equal(summed.bytes, 21_000)
+  assert.equal(summed.conversations, 2)
+  const deduped = buildFlowAnalytics({ minutes: 5, serverId: 7, dedup: true })
+  assert.equal(deduped.bytes, 12_000)
+  assert.equal(deduped.conversations, 1)
+  assert.equal(deduped.dedupApplied, true)
+  assert.equal(deduped.interfaces.length, 2)
+} finally {
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+}
+
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+disableRipeEnqueueForTests()
+seedRipeCacheForTests({
+  prefix: "8.8.8.0/24",
+  asn: 15169,
+  country: "US",
+  lat: 37.4,
+  lng: -122.1,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+})
+seedFlowCatalogForTests({
+  cidrs: [{ cidr: "8.8.8.0/24", purpose: "steam-gaming" }],
+})
+rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
+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: "",
+  },
+])
+try {
+  const geo = buildFlowAnalytics({ minutes: 5, serverId: 7 })
+  assert.equal(geo.categories?.[0]?.label, "Игры")
+  assert.ok(geo.asns?.some((r) => r.label.includes("AS15169")))
+  assert.equal(geo.countries?.[0]?.id, "US")
+  assert.equal(geo.mapEdges?.[0]?.toCountry, "US")
+  assert.equal(geo.conversationsList[0]?.dstCountry, "US")
+} finally {
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
 console.log("traffic-flow-analytics.test.ts: ok")

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

@@ -7,6 +7,7 @@ import type {
   FlowClientsDto,
   FlowEntityCard,
   FlowExportersDto,
+  FlowMapEdge,
   FlowTalkerDto,
 } from "@mmapp/contracts/traffic-flow"
 import { protoName } from "./traffic-flow-parse.js"
@@ -19,12 +20,17 @@ import {
 import { resolveIfaceName } from "./traffic-flow-ifaces.js"
 import { getTrafficFlowSettingsRow, listHostPeers } from "./traffic-flow-settings.js"
 import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
+import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js"
+import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js"
+import { classifyFlowDst, refreshFlowCatalogInBackground } from "./traffic-flow-classify.js"
 
 export interface FlowAnalyticsQuery {
   minutes: number
   serverId?: number
   userId?: string
   iface?: string
+  /** Default true: один 5-tuple = max байт по ifaces. */
+  dedup?: boolean
 }
 
 function bpsToMbps(bps: number): number {
@@ -89,6 +95,18 @@ function snapshotStatus(serverId: number): FlowEntityCard["status"] {
   return "online"
 }
 
+function topLabel(map: Map<string, { bytes: number; packets: number }>, fallback = "—"): string {
+  let best = fallback
+  let bestBytes = 0
+  for (const [label, v] of map) {
+    if (v.bytes > bestBytes) {
+      bestBytes = v.bytes
+      best = label
+    }
+  }
+  return best
+}
+
 export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
   const settings = getTrafficFlowSettingsRow()
   const top = Math.min(50, Math.max(10, settings.topN))
@@ -98,39 +116,70 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
   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 countryById = new Map(serverRows.map((s) => [s.id, (s.country || "").toUpperCase() || "UN"]))
+  const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : ""
+  const wantDedup = q.dedup !== false && !ifaceFilter
+
+  refreshFlowCatalogInBackground()
 
   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 asns = new Map<string, { bytes: number; packets: number }>()
+  const countries = new Map<string, { bytes: number; packets: number }>()
+  const categories = new Map<string, { bytes: number; packets: number }>()
+  const services = new Map<string, { bytes: number; packets: number }>()
   const conv = new Map<string, FlowTalkerDto & { rawBytes: number }>()
+  const edgeAcc = new Map<string, FlowMapEdge & { catBytes: Map<string, 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 working = wantDedup ? dedupFlowRowsMaxBytes(matched) : matched
+  const conversationsRaw = new Set(matched.map((r) => `${flowTupleKey(r)}|${r.inIface}`)).size
+
+  let totalBytes = 0
+  let totalPackets = 0
+  for (const r of working) {
+    const resolved = resolveIfaceName(r.serverId, r.inIface)
+    totalBytes += r.bytes
+    totalPackets += r.packets
+    srcs.add(r.src)
+    dsts.add(r.dst)
+    const app = applicationName(r.proto, r.dstPort, r.srcPort)
+    const ripe = lookupRipeCached(r.dst)
+    const classified = classifyFlowDst(r.dst, r.proto, r.dstPort, r.srcPort, ripe)
+    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)
+    bump(categories, classified.category, r.bytes, r.packets)
+    bump(services, classified.service, r.bytes, r.packets)
+    if (ripe?.ok && ripe.asn) {
+      const asnLabel = ripe.holder ? `AS${ripe.asn} ${ripe.holder}` : `AS${ripe.asn}`
+      bump(asns, asnLabel, r.bytes, r.packets)
+    }
+    if (ripe?.ok && ripe.country && ripe.country !== "—") {
+      bump(countries, ripe.country, r.bytes, r.packets)
+    }
+
+    const ckey = wantDedup
+      ? flowTupleKey(r)
+      : `${flowTupleKey(r)}|${r.inIface}`
     const prev = conv.get(ckey)
     if (prev) {
       prev.rawBytes += r.bytes
@@ -152,30 +201,53 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
         inIface: resolved.name,
         inIfaceIndex: resolved.index,
         application: app,
+        category: classified.category,
+        service: classified.service,
+        dstCountry: ripe?.country && ripe.country !== "—" ? ripe.country : undefined,
+        dstAsn: ripe?.asn || undefined,
         rawBytes: r.bytes,
       })
     }
+
+    const toCountry = ripe?.ok && ripe.country && ripe.country !== "—" ? ripe.country : ""
+    if (toCountry) {
+      const fromCountry = countryById.get(r.serverId) || "UN"
+      const ekey = `${r.serverId}|${toCountry}`
+      let edge = edgeAcc.get(ekey)
+      if (!edge) {
+        edge = {
+          fromId: String(r.serverId),
+          fromLabel: nameById.get(r.serverId) ?? String(r.serverId),
+          fromCountry,
+          toCountry,
+          toAsn: ripe?.asn ?? 0,
+          category: classified.category,
+          bytes: 0,
+          bps: 0,
+          catBytes: new Map(),
+        }
+        edgeAcc.set(ekey, edge)
+      }
+      edge.bytes += r.bytes
+      if (ripe?.asn) edge.toAsn = ripe.asn
+      edge.catBytes.set(classified.category, (edge.catBytes.get(classified.category) ?? 0) + r.bytes)
+    }
   }
 
+  enqueueRipeMisses(dsts)
+
   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 topProto = topLabel(protocols)
+  const topCategory = topLabel(categories)
 
-  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))
+    ? getRingMbps(ringServer, ifaceFilter === "" ? "__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)
@@ -190,21 +262,46 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
       bps: (v.bytes * 8) / windowSec,
     }))
 
-  const protoBreakdown = topN(protocols, windowSec, top)
+  const ifaceRawBytes = [...ifacesMap.values()].reduce((a, v) => a + v.bytes, 0) || 1
   const listener = getFlowListenerState()
+  const mapEdges: FlowMapEdge[] = [...edgeAcc.values()]
+    .map((e) => {
+      let cat = e.category
+      let catBest = 0
+      for (const [label, bytes] of e.catBytes) {
+        if (bytes > catBest) {
+          catBest = bytes
+          cat = label
+        }
+      }
+      return {
+        fromId: e.fromId,
+        fromLabel: e.fromLabel,
+        fromCountry: e.fromCountry,
+        toCountry: e.toCountry,
+        toAsn: e.toAsn,
+        category: cat,
+        bytes: e.bytes,
+        bps: (e.bytes * 8) / windowSec,
+      }
+    })
+    .sort((a, b) => b.bytes - a.bytes)
+    .slice(0, top)
 
   return {
     bpsNow: (ring.rxNow + ring.txNow) * 1_000_000 || (totalBytes * 8) / windowSec,
     bytes: totalBytes,
     packets: totalPackets,
     conversations: conv.size,
+    conversationsRaw,
     uniqueSrc: srcs.size,
     uniqueDst: dsts.size,
     topProto,
+    topCategory,
     rxSeries,
     txSeries,
     applications: topN(applications, windowSec, top),
-    protocols: protoBreakdown,
+    protocols: topN(protocols, windowSec, top),
     sources: topN(sources, windowSec, top),
     destinations: topN(destinations, windowSec, top),
     interfaces: [...ifacesMap.entries()].map(([label, v]) => ({
@@ -213,11 +310,17 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
       bytes: v.bytes,
       packets: v.packets,
       bps: (v.bytes * 8) / windowSec,
-      percent: totalBytes > 0 ? (v.bytes / totalBytes) * 100 : 0,
+      percent: (v.bytes / ifaceRawBytes) * 100,
     })).sort((a, b) => b.bytes - a.bytes),
+    asns: topN(asns, windowSec, top),
+    countries: topN(countries, windowSec, top),
+    categories: topN(categories, windowSec, top),
+    services: topN(services, windowSec, top),
+    mapEdges,
     conversationsList,
     ifaces: ifaceRows,
     live: listener.bound,
+    dedupApplied: wantDedup,
   }
 }
 

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

@@ -0,0 +1,20 @@
+import assert from "node:assert/strict"
+import { classifyFlowDst, disableCatalogFetchForTests, resetFlowCatalogForTests, seedFlowCatalogForTests } from "./traffic-flow-classify.js"
+
+disableCatalogFetchForTests()
+resetFlowCatalogForTests()
+seedFlowCatalogForTests({
+  cidrs: [{ cidr: "192.0.2.0/24", purpose: "steam-gaming" }],
+})
+
+const hit = classifyFlowDst("192.0.2.10", 6, 443, 50000, null)
+assert.equal(hit.category, "Игры")
+assert.equal(hit.service, "steam-gaming")
+
+const miss = classifyFlowDst("203.0.113.9", 17, 53, 53000, null)
+assert.equal(miss.category, "DNS")
+
+const cdn = classifyFlowDst("203.0.113.9", 6, 443, 1, { prefix: "203.0.113.0/24", asn: 13335, country: "US", lat: null, lng: null, holder: "CLOUDFLARENET", ok: true, fetchedAt: Date.now() })
+assert.equal(cdn.category, "CDN")
+
+console.log("traffic-flow-classify.test.ts: ok")

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

@@ -0,0 +1,140 @@
+import { db } from "../db/index.js"
+import { evobgpSettings } from "../db/schema.js"
+import { ipInCidrV4, parseCidrV4 } from "./traffic-flow-ip.js"
+import type { FlowIpMeta } from "./traffic-flow-ripe.js"
+import { applicationName } from "./traffic-flow-apps.js"
+
+export interface FlowClassification {
+  service: string
+  category: string
+}
+
+interface CatalogCidr {
+  cidr: string
+  purpose: string
+  prefixLen: number
+}
+
+const CATALOG_TTL_MS = 10 * 60_000
+let cidrs: CatalogCidr[] = []
+let asnPurpose = new Map<number, string>()
+let fetchedAt = 0
+let catalogFetchEnabled = true
+let inflight: Promise<void> | null = null
+
+export function disableCatalogFetchForTests(): void {
+  catalogFetchEnabled = false
+}
+
+export function resetFlowCatalogForTests(): void {
+  cidrs = []
+  asnPurpose = new Map()
+  fetchedAt = 0
+  inflight = null
+}
+
+export function seedFlowCatalogForTests(input: {
+  cidrs?: Array<{ cidr: string; purpose: string }>
+  asns?: Array<{ asn: number; purpose: string }>
+}): void {
+  cidrs = (input.cidrs ?? [])
+    .map((c) => ({ cidr: c.cidr, purpose: c.purpose, prefixLen: parseCidrV4(c.cidr)?.prefixLen ?? 0 }))
+    .sort((a, b) => b.prefixLen - a.prefixLen)
+  asnPurpose = new Map((input.asns ?? []).map((a) => [a.asn, a.purpose]))
+  fetchedAt = Date.now()
+}
+
+export function categoryFromPurpose(purpose: string, proto: number, dstPort: number, srcPort: number): string {
+  const p = purpose.toLowerCase()
+  if (/gaming|steam|epic|riot/.test(p)) return "Игры"
+  if (/streaming|youtube|netflix|twitch|video/.test(p)) return "Видео / стриминг"
+  if (/cdn|cloudflare|akamai|fastly/.test(p)) return "CDN"
+  if (/voip|discord|zoom/.test(p)) return "Голос"
+  const app = applicationName(proto, dstPort, srcPort)
+  if (app === "DNS" || app === "SSH" || app === "BGP") return app
+  return "Проче"
+}
+
+function matchCidr(ip: string): CatalogCidr | null {
+  for (const row of cidrs) {
+    if (ipInCidrV4(ip, row.cidr)) return row
+  }
+  return null
+}
+
+export function classifyFlowDst(
+  dst: string,
+  proto: number,
+  dstPort: number,
+  srcPort: number,
+  ripe: FlowIpMeta | null,
+): FlowClassification {
+  const hit = matchCidr(dst)
+  const asnName = ripe?.asn ? asnPurpose.get(ripe.asn) : undefined
+  const purpose = hit?.purpose || asnName || ripe?.holder || ""
+  const service = (hit?.purpose || asnName || ripe?.holder || "Проче").trim() || "Проче"
+  return {
+    service,
+    category: categoryFromPurpose(purpose, proto, dstPort, srcPort),
+  }
+}
+
+async function fetchCatalog(): Promise<void> {
+  if (!catalogFetchEnabled) return
+  if (Date.now() - fetchedAt < CATALOG_TTL_MS) return
+  if (inflight) return inflight
+  inflight = (async () => {
+    try {
+      const row = db.select().from(evobgpSettings).limit(1).all()[0]
+      if (!row?.enabled) return
+      const root = String(row.baseUrl ?? "").replace(/\/+$/, "")
+      const token = String(row.apiKey ?? "").replace(/^Bearer\s+/i, "").trim()
+      if (!root || !token) return
+      const ac = new AbortController()
+      const t = setTimeout(() => ac.abort(), 20_000)
+      try {
+        const res = await fetch(`${root}/v1/router-lists/catalog`, {
+          headers: { Authorization: `Bearer ${token}`, Accept: "application/json" },
+          signal: ac.signal,
+        })
+        if (!res.ok) return
+        const catalog = await res.json() as {
+          modules?: { items?: Array<{ id: string; name: string }> }
+          ip_ranges?: { items?: Array<{ module_id: string; entry: { prefix: string } }> }
+          asns?: { items?: Array<{ module_id: string; entry: { asn: number } }> }
+        }
+        const mods = new Map((catalog.modules?.items ?? []).map((m) => [m.id, m.name]))
+        const next: CatalogCidr[] = []
+        for (const item of catalog.ip_ranges?.items ?? []) {
+          const prefix = String(item.entry?.prefix ?? "").trim()
+          const purpose = mods.get(item.module_id) ?? ""
+          const parsed = parseCidrV4(prefix)
+          if (!prefix || !parsed) continue
+          next.push({ cidr: prefix, purpose, prefixLen: parsed.prefixLen })
+        }
+        next.sort((a, b) => b.prefixLen - a.prefixLen)
+        const nextAsn = new Map<number, string>()
+        for (const item of catalog.asns?.items ?? []) {
+          const purpose = mods.get(item.module_id)
+          const asn = Number(item.entry?.asn)
+          if (purpose && Number.isFinite(asn) && asn > 0) nextAsn.set(asn, purpose)
+        }
+        cidrs = next
+        asnPurpose = nextAsn
+        fetchedAt = Date.now()
+      } finally {
+        clearTimeout(t)
+      }
+    } catch {
+      /* catalog optional */
+    } finally {
+      inflight = null
+    }
+  })()
+  return inflight
+}
+
+/** Background refresh — analytics never awaits the HTTP. */
+export function refreshFlowCatalogInBackground(): void {
+  void fetchCatalog()
+}

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

@@ -0,0 +1,25 @@
+import assert from "node:assert/strict"
+import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js"
+
+const a = {
+  serverId: 7,
+  src: "10.1.1.8",
+  dst: "8.8.8.8",
+  proto: 6,
+  srcPort: 1,
+  dstPort: 443,
+  inIface: "2",
+  bytes: 12_000,
+  packets: 10,
+}
+const b = { ...a, inIface: "10", bytes: 8_000, packets: 8 }
+const out = dedupFlowRowsMaxBytes([a, b])
+assert.equal(out.length, 1)
+assert.equal(out[0]?.bytes, 12_000)
+assert.equal(out[0]?.inIface, "2")
+assert.equal(flowTupleKey(a), flowTupleKey(b))
+
+const sameIface = dedupFlowRowsMaxBytes([a, { ...a, bytes: 3_000, packets: 2 }])
+assert.equal(sameIface[0]?.bytes, 15_000)
+
+console.log("traffic-flow-dedup.test.ts: ok")

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

@@ -0,0 +1,44 @@
+export interface FlowTupleRow {
+  serverId: number
+  src: string
+  dst: string
+  proto: number
+  srcPort: number
+  dstPort: number
+  inIface: string
+  bytes: number
+  packets: number
+}
+
+export function flowTupleKey(r: Pick<FlowTupleRow, "serverId" | "src" | "dst" | "proto" | "srcPort" | "dstPort">): string {
+  return `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}`
+}
+
+function ifaceKey(r: FlowTupleRow): string {
+  return `${flowTupleKey(r)}|${r.inIface}`
+}
+
+/**
+ * Один 5-tuple на двух ifIndex — это один поток: сначала сумма по бакетам/iface,
+ * затем max байт между интерфейсами (не sum).
+ */
+export function dedupFlowRowsMaxBytes<T extends FlowTupleRow>(rows: T[]): T[] {
+  const byIface = new Map<string, T>()
+  for (const row of rows) {
+    const key = ifaceKey(row)
+    const prev = byIface.get(key)
+    if (!prev) {
+      byIface.set(key, { ...row })
+      continue
+    }
+    prev.bytes += row.bytes
+    prev.packets += row.packets
+  }
+  const byTuple = new Map<string, T>()
+  for (const row of byIface.values()) {
+    const key = flowTupleKey(row)
+    const prev = byTuple.get(key)
+    if (!prev || row.bytes > prev.bytes) byTuple.set(key, row)
+  }
+  return [...byTuple.values()]
+}

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

@@ -27,6 +27,14 @@ assert.equal(resolveIfaceName(7, "0").name, "—")
 assert.equal(resolveIfaceName(7, "ether1").name, "ether1")
 assert.equal(resolveIfaceName(7, "99").name, "#99")
 
+resetIfaceCacheForTests()
+rememberServerIfaces(8, [
+  { ifindex: "10", ".id": "*12", name: "gre1" },
+])
+assert.equal(rosIdToIfIndex("*12"), 18)
+assert.equal(resolveIfaceName(8, "10").name, "gre1")
+assert.equal(resolveIfaceName(8, "18").name, "gre1")
+
 assert.equal(applicationName(6, 443), "HTTPS")
 assert.equal(applicationName(17, 53), "DNS")
 assert.equal(applicationName(6, 22), "SSH")

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

@@ -9,6 +9,7 @@ import {
 } from "./traffic-flow-ifindex.js"
 
 export {
+  ifaceCacheFresh,
   ifaceCacheHas,
   rememberServerIfaces,
   resetIfaceCacheForTests,

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

@@ -25,10 +25,9 @@ export function rememberServerIfaces(serverId: number, rows: RosIfaceIndexRow[])
     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)
+    const fromId = rosIdToIfIndex(row[".id"])
+    if (Number.isFinite(fromProp) && fromProp > 0) map.set(fromProp, name)
+    if (fromId != null && fromId > 0) map.set(fromId, name)
   }
   cache.set(serverId, map)
   fetchedAt.set(serverId, Date.now())

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

@@ -11,7 +11,7 @@ import {
   recordFlowListenerError,
   recordFlowPacket,
 } from "./traffic-flow-settings.js"
-import { ifaceCacheHas, refreshServerIfaces, resolveIfaceName } from "./traffic-flow-ifaces.js"
+import { ifaceCacheFresh, refreshServerIfaces, resolveIfaceName } from "./traffic-flow-ifaces.js"
 import { applicationName } from "./traffic-flow-apps.js"
 
 export interface FlowListenerState {
@@ -145,7 +145,9 @@ function resolveServerId(exporterIp: string): number | null {
 function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean {
   const serverId = resolveServerId(exporterIp)
   if (serverId == null) return false
-  if (!ifaceCacheHas(serverId)) void refreshServerIfaces(serverId)
+  const needsRefresh = !ifaceCacheFresh(serverId)
+    || flows.some((f) => resolveIfaceName(serverId, f.inIface).name.startsWith("#"))
+  if (needsRefresh) void refreshServerIfaces(serverId, true)
   const bucketAt = minuteBucketIso()
   for (const flow of flows) {
     addToTick(serverId, flow.inIface, flow.outIface, flow.bytes)

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

@@ -0,0 +1,54 @@
+/** IPv4 helpers for RIPEstat prefix cache and EvoBGP CIDR match. */
+
+export function ipv4ToInt(ip: string): number | null {
+  const parts = String(ip ?? "").trim().split(".")
+  if (parts.length !== 4) return null
+  let n = 0
+  for (const p of parts) {
+    if (!/^\d+$/.test(p)) return null
+    const o = Number(p)
+    if (o < 0 || o > 255) return null
+    n = ((n << 8) >>> 0) + o
+  }
+  return n >>> 0
+}
+
+export function parseCidrV4(cidr: string): { net: number; mask: number; prefixLen: number } | null {
+  const raw = String(cidr ?? "").trim()
+  const [ip, lenRaw] = raw.split("/")
+  const addr = ipv4ToInt(ip ?? "")
+  const prefixLen = Number.parseInt(lenRaw ?? "", 10)
+  if (addr == null || !Number.isFinite(prefixLen) || prefixLen < 0 || prefixLen > 32) return null
+  const mask = prefixLen === 0 ? 0 : (0xffffffff << (32 - prefixLen)) >>> 0
+  return { net: (addr & mask) >>> 0, mask, prefixLen }
+}
+
+export function ipInCidrV4(ip: string, cidr: string): boolean {
+  const addr = ipv4ToInt(ip)
+  const parsed = parseCidrV4(cidr)
+  if (addr == null || !parsed) return false
+  return ((addr & parsed.mask) >>> 0) === parsed.net
+}
+
+export function isNonPublicIp(ip: string): boolean {
+  const trimmed = String(ip ?? "").trim()
+  if (!trimmed) return true
+  if (trimmed.includes(":")) {
+    const lower = trimmed.toLowerCase()
+    return lower === "::1" || lower.startsWith("fe80:") || lower.startsWith("fc") || lower.startsWith("fd") || lower === "::"
+  }
+  const n = ipv4ToInt(trimmed)
+  if (n == null) return true
+  const inRange = (cidr: string) => ipInCidrV4(trimmed, cidr)
+  return (
+    inRange("0.0.0.0/8")
+    || inRange("10.0.0.0/8")
+    || inRange("127.0.0.0/8")
+    || inRange("169.254.0.0/16")
+    || inRange("172.16.0.0/12")
+    || inRange("192.168.0.0/16")
+    || inRange("100.64.0.0/10")
+    || inRange("224.0.0.0/4")
+    || inRange("255.255.255.255/32")
+  )
+}

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

@@ -96,7 +96,7 @@ resetFlowTemplatesForTests()
   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]?.inIface, "ether1")
   assert.equal(named[0]?.src, "10.1.1.8")
 }
 

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

@@ -206,7 +206,7 @@ function recordFromFields(
     }
     off = field.next
   }
-  if (!inIface && ifaceName) inIface = ifaceName
+  if (ifaceName) inIface = ifaceName
   return { flow: { src, dst, proto, srcPort, dstPort, bytes, packets, inIface, outIface }, next: off }
 }
 

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

@@ -0,0 +1,72 @@
+import assert from "node:assert/strict"
+import {
+  disableRipeEnqueueForTests,
+  disableRipePersistForTests,
+  enqueueRipeMisses,
+  flushRipeQueueForTests,
+  lookupRipeCached,
+  resetRipeCacheForTests,
+  ripeFetchCountForTests,
+  seedRipeCacheForTests,
+  setRipeFetchForTests,
+} from "./traffic-flow-ripe.js"
+
+disableRipePersistForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+
+assert.equal(lookupRipeCached("10.1.1.8")?.ok, false)
+assert.equal(lookupRipeCached("192.168.0.1")?.ok, false)
+assert.equal(lookupRipeCached("100.64.1.2")?.ok, false)
+assert.equal(ripeFetchCountForTests(), 0)
+
+seedRipeCacheForTests({
+  prefix: "1.2.3.0/24",
+  asn: 64500,
+  country: "NL",
+  lat: 52.3,
+  lng: 4.9,
+  holder: "TEST",
+  ok: true,
+  fetchedAt: Date.now(),
+})
+assert.equal(lookupRipeCached("1.2.3.10")?.country, "NL")
+assert.equal(lookupRipeCached("1.2.3.10")?.asn, 64500)
+assert.equal(ripeFetchCountForTests(), 0)
+
+resetRipeCacheForTests()
+disableRipePersistForTests()
+setRipeFetchForTests(async (input) => {
+  const url = String(input)
+  const body = url.includes("network-info")
+    ? { data: { prefix: "8.8.8.0/24", asns: ["15169"] } }
+    : url.includes("maxmind-geo-lite")
+      ? { data: { located_resources: [{ locations: [{ country: "US", latitude: 37.4, longitude: -122.1 }] }] } }
+      : { data: { holder: "GOOGLE" } }
+  return new Response(JSON.stringify(body), { status: 200, headers: { "Content-Type": "application/json" } })
+})
+enqueueRipeMisses(["8.8.8.8"])
+await flushRipeQueueForTests()
+assert.equal(lookupRipeCached("8.8.8.8")?.country, "US")
+assert.equal(lookupRipeCached("8.8.8.10")?.prefix, "8.8.8.0/24")
+const afterFirst = ripeFetchCountForTests()
+assert.ok(afterFirst >= 2)
+enqueueRipeMisses(["8.8.8.10"])
+await flushRipeQueueForTests()
+assert.equal(ripeFetchCountForTests(), afterFirst)
+
+resetRipeCacheForTests()
+disableRipePersistForTests()
+setRipeFetchForTests(async () => {
+  throw new Error("timeout")
+})
+enqueueRipeMisses(["203.0.113.50"])
+await flushRipeQueueForTests()
+const neg = lookupRipeCached("203.0.113.50")
+assert.equal(neg?.ok, false)
+const afterNeg = ripeFetchCountForTests()
+enqueueRipeMisses(["203.0.113.50"])
+await flushRipeQueueForTests()
+assert.equal(ripeFetchCountForTests(), afterNeg)
+
+console.log("traffic-flow-ripe.test.ts: ok")

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

@@ -0,0 +1,371 @@
+import { sqliteDatabase } from "../db/index.js"
+import { ipInCidrV4, ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
+
+export interface FlowIpMeta {
+  prefix: string
+  asn: number
+  country: string
+  lat: number | null
+  lng: number | null
+  holder: string
+  ok: boolean
+  fetchedAt: number
+}
+
+const HIT_TTL_MS = 24 * 60 * 60_000
+const NEG_TTL_MS = 6 * 60 * 60_000
+const MAX_NEW_PREFIX_PER_MIN = 30
+const CONCURRENCY = 3
+const RIPE_BASE = "https://stat.ripe.net/data"
+const UA = "MikrotikManager-flow/1.0"
+
+const mem = new Map<string, FlowIpMeta>()
+const asnHolder = new Map<number, { holder: string; fetchedAt: number }>()
+const inflight = new Map<string, Promise<FlowIpMeta | null>>()
+const queue: string[] = []
+const queued = new Set<string>()
+const recentFetches: number[] = []
+
+let persistEnabled = true
+let enqueueEnabled = true
+let loaded = false
+let workerRunning = false
+let fetchImpl: typeof fetch = globalThis.fetch.bind(globalThis)
+let fetchCount = 0
+
+export function disableRipePersistForTests(): void {
+  persistEnabled = false
+}
+
+export function disableRipeEnqueueForTests(): void {
+  enqueueEnabled = false
+}
+
+export function resetRipeCacheForTests(): void {
+  mem.clear()
+  asnHolder.clear()
+  inflight.clear()
+  queue.length = 0
+  queued.clear()
+  recentFetches.length = 0
+  loaded = persistEnabled ? false : true
+  workerRunning = false
+  fetchCount = 0
+  enqueueEnabled = true
+  fetchImpl = globalThis.fetch.bind(globalThis)
+}
+
+export function seedRipeCacheForTests(entry: FlowIpMeta): void {
+  mem.set(entry.prefix, { ...entry })
+  loaded = true
+}
+
+export function setRipeFetchForTests(fn: typeof fetch): void {
+  fetchImpl = fn
+  fetchCount = 0
+}
+
+export function ripeFetchCountForTests(): number {
+  return fetchCount
+}
+
+export async function flushRipeQueueForTests(timeoutMs = 4000): Promise<void> {
+  const start = Date.now()
+  while (Date.now() - start < timeoutMs) {
+    if (!queue.length && !inflight.size && !workerRunning) return
+    await new Promise((r) => setTimeout(r, 20))
+  }
+}
+
+function ttlMs(ok: boolean): number {
+  return ok ? HIT_TTL_MS : NEG_TTL_MS
+}
+
+function isFresh(entry: FlowIpMeta): boolean {
+  return Date.now() - entry.fetchedAt < ttlMs(entry.ok)
+}
+
+function loadSqlite(): void {
+  if (loaded || !persistEnabled) {
+    loaded = true
+    return
+  }
+  loaded = true
+  try {
+    const rows = sqliteDatabase.prepare(`
+      SELECT prefix, asn, country, lat, lng, holder, ok, fetched_at
+      FROM flow_ip_meta
+    `).all() as Array<{
+      prefix: string
+      asn: number | null
+      country: string
+      lat: number | null
+      lng: number | null
+      holder: string
+      ok: number
+      fetched_at: string
+    }>
+    for (const r of rows) {
+      const fetchedAt = Date.parse(r.fetched_at)
+      mem.set(r.prefix, {
+        prefix: r.prefix,
+        asn: Number(r.asn ?? 0) || 0,
+        country: r.country || "—",
+        lat: r.lat == null ? null : Number(r.lat),
+        lng: r.lng == null ? null : Number(r.lng),
+        holder: r.holder || "",
+        ok: r.ok !== 0,
+        fetchedAt: Number.isFinite(fetchedAt) ? fetchedAt : 0,
+      })
+    }
+    const asns = sqliteDatabase.prepare(`SELECT asn, holder, fetched_at FROM flow_asn_meta`).all() as Array<{
+      asn: number
+      holder: string
+      fetched_at: string
+    }>
+    for (const a of asns) {
+      const fetchedAt = Date.parse(a.fetched_at)
+      asnHolder.set(a.asn, { holder: a.holder || "", fetchedAt: Number.isFinite(fetchedAt) ? fetchedAt : 0 })
+    }
+  } catch {
+    /* table may not exist in isolated tests */
+  }
+}
+
+function persist(entry: FlowIpMeta): void {
+  if (!persistEnabled) return
+  try {
+    sqliteDatabase.prepare(`
+      INSERT INTO flow_ip_meta (prefix, asn, country, lat, lng, holder, ok, fetched_at)
+      VALUES (@prefix, @asn, @country, @lat, @lng, @holder, @ok, @fetchedAt)
+      ON CONFLICT(prefix) DO UPDATE SET
+        asn=excluded.asn, country=excluded.country, lat=excluded.lat, lng=excluded.lng,
+        holder=excluded.holder, ok=excluded.ok, fetched_at=excluded.fetched_at
+    `).run({
+      prefix: entry.prefix,
+      asn: entry.asn,
+      country: entry.country,
+      lat: entry.lat,
+      lng: entry.lng,
+      holder: entry.holder,
+      ok: entry.ok ? 1 : 0,
+      fetchedAt: new Date(entry.fetchedAt).toISOString(),
+    })
+  } catch {
+    /* ignore persist errors */
+  }
+}
+
+function persistAsn(asn: number, holder: string): void {
+  if (!persistEnabled || !asn) return
+  try {
+    sqliteDatabase.prepare(`
+      INSERT INTO flow_asn_meta (asn, holder, fetched_at)
+      VALUES (@asn, @holder, @fetchedAt)
+      ON CONFLICT(asn) DO UPDATE SET holder=excluded.holder, fetched_at=excluded.fetched_at
+    `).run({
+      asn,
+      holder,
+      fetchedAt: new Date().toISOString(),
+    })
+  } catch {
+    /* ignore */
+  }
+}
+
+function negative(prefix: string): FlowIpMeta {
+  return {
+    prefix,
+    asn: 0,
+    country: "—",
+    lat: null,
+    lng: null,
+    holder: "",
+    ok: false,
+    fetchedAt: Date.now(),
+  }
+}
+
+export function lookupRipeCached(ip: string): FlowIpMeta | null {
+  loadSqlite()
+  const trimmed = String(ip ?? "").trim()
+  if (!trimmed) return null
+  if (isNonPublicIp(trimmed)) {
+    return negative(`${trimmed.includes(":") ? trimmed : trimmed}/32`)
+  }
+  let best: FlowIpMeta | null = null
+  let bestLen = -1
+  for (const entry of mem.values()) {
+    if (!isFresh(entry)) continue
+    const parsed = parseCidrV4(entry.prefix)
+    if (!parsed) continue
+    if (!ipInCidrV4(trimmed, entry.prefix)) continue
+    if (parsed.prefixLen > bestLen) {
+      best = entry
+      bestLen = parsed.prefixLen
+    }
+  }
+  return best
+}
+
+async function ripeJson(path: string, resource: string): Promise<unknown> {
+  fetchCount += 1
+  const url = `${RIPE_BASE}/${path}/data.json?resource=${encodeURIComponent(resource)}`
+  const ac = new AbortController()
+  const t = setTimeout(() => ac.abort(), 12_000)
+  try {
+    const res = await fetchImpl(url, {
+      headers: { Accept: "application/json", "User-Agent": UA },
+      signal: ac.signal,
+    })
+    if (!res.ok) throw new Error(`HTTP ${res.status}`)
+    return await res.json()
+  } finally {
+    clearTimeout(t)
+  }
+}
+
+function pickPrefix(data: unknown): string {
+  const d = data as { data?: { prefix?: string } }
+  return String(d?.data?.prefix ?? "").trim()
+}
+
+function pickAsns(data: unknown): number {
+  const d = data as { data?: { asns?: unknown } }
+  const raw = d?.data?.asns
+  const first = Array.isArray(raw) ? raw[0] : raw
+  const n = Number.parseInt(String(first ?? "").replace(/^AS/i, ""), 10)
+  return Number.isFinite(n) ? n : 0
+}
+
+function pickGeo(data: unknown): { country: string; lat: number | null; lng: number | null } {
+  const d = data as {
+    data?: {
+      located_resources?: Array<{
+        locations?: Array<{ country?: string; latitude?: number; longitude?: number }>
+      }>
+    }
+  }
+  const loc = d?.data?.located_resources?.[0]?.locations?.[0]
+  const country = String(loc?.country ?? "").trim().toUpperCase()
+  const lat = loc?.latitude == null ? null : Number(loc.latitude)
+  const lng = loc?.longitude == null ? null : Number(loc.longitude)
+  return {
+    country: country || "—",
+    lat: Number.isFinite(lat) ? lat : null,
+    lng: Number.isFinite(lng) ? lng : null,
+  }
+}
+
+function pickHolder(data: unknown): string {
+  const d = data as { data?: { holder?: string } }
+  return String(d?.data?.holder ?? "").trim()
+}
+
+function allowNewPrefix(): boolean {
+  const now = Date.now()
+  while (recentFetches.length && now - recentFetches[0]! > 60_000) recentFetches.shift()
+  return recentFetches.length < MAX_NEW_PREFIX_PER_MIN
+}
+
+async function resolveIp(ip: string): Promise<FlowIpMeta | null> {
+  const cached = lookupRipeCached(ip)
+  if (cached) return cached
+  const pending = inflight.get(ip)
+  if (pending) return pending
+
+  const job = (async () => {
+    if (!allowNewPrefix()) return null
+    recentFetches.push(Date.now())
+    try {
+      const net = await ripeJson("network-info", ip)
+      const prefix = pickPrefix(net) || `${ip}/32`
+      const existing = mem.get(prefix)
+      if (existing && isFresh(existing)) return existing
+      const asn = pickAsns(net)
+      let geo = { country: "—", lat: null as number | null, lng: null as number | null }
+      try {
+        geo = pickGeo(await ripeJson("maxmind-geo-lite", prefix))
+      } catch {
+        /* best-effort */
+      }
+      let holder = asnHolder.get(asn)?.holder ?? ""
+      if (asn && (!holder || Date.now() - (asnHolder.get(asn)?.fetchedAt ?? 0) > HIT_TTL_MS)) {
+        try {
+          holder = pickHolder(await ripeJson("as-overview", `AS${asn}`))
+          asnHolder.set(asn, { holder, fetchedAt: Date.now() })
+          persistAsn(asn, holder)
+        } catch {
+          /* best-effort */
+        }
+      }
+      const entry: FlowIpMeta = {
+        prefix,
+        asn,
+        country: geo.country,
+        lat: geo.lat,
+        lng: geo.lng,
+        holder,
+        ok: Boolean(asn || (geo.country && geo.country !== "—")),
+        fetchedAt: Date.now(),
+      }
+      mem.set(prefix, entry)
+      persist(entry)
+      return entry
+    } catch {
+      const prefix = `${ip}/32`
+      const entry = negative(prefix)
+      mem.set(prefix, entry)
+      persist(entry)
+      return entry
+    } finally {
+      inflight.delete(ip)
+    }
+  })()
+
+  inflight.set(ip, job)
+  return job
+}
+
+async function runWorker(): Promise<void> {
+  if (workerRunning) return
+  workerRunning = true
+  try {
+    while (queue.length) {
+      const batch: string[] = []
+      while (batch.length < CONCURRENCY && queue.length) {
+        const ip = queue.shift()
+        if (!ip) break
+        queued.delete(ip)
+        if (lookupRipeCached(ip)) continue
+        if (ipv4ToInt(ip) == null && !ip.includes(":")) continue
+        batch.push(ip)
+      }
+      if (!batch.length) {
+        if (!allowNewPrefix()) {
+          await new Promise((r) => setTimeout(r, 1000))
+        }
+        continue
+      }
+      await Promise.all(batch.map((ip) => resolveIp(ip)))
+    }
+  } finally {
+    workerRunning = false
+    if (queue.length) void runWorker()
+  }
+}
+
+/** HTTP / SSE never await this — cache miss is filled on a later tick. */
+export function enqueueRipeMisses(ips: Iterable<string>): void {
+  if (!enqueueEnabled) return
+  loadSqlite()
+  for (const raw of ips) {
+    const ip = String(raw ?? "").trim()
+    if (!ip || isNonPublicIp(ip)) continue
+    if (lookupRipeCached(ip)) continue
+    if (queued.has(ip) || inflight.has(ip)) continue
+    queued.add(ip)
+    queue.push(ip)
+  }
+  if (queue.length) void runWorker()
+}