feat(traffic-flow): enhance country handling in flow analytics

- Introduced country-specific handling in traffic flow analytics, including new mappings for country nodes and edges.
- Added mock data for countries and their respective service paths to improve visualization in the Network Map.
- Updated the `countryName` function to utilize `Intl.DisplayNames` for better localization of country names.
- Enhanced tests to validate country handling and ensure accurate representation in flow analytics.

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

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

@@ -8,6 +8,7 @@ import {
   OTHER_SERVICE,
   isNamedInternetService,
   mapServiceNodeId,
+  mapCountryNodeId,
   resolveFlowBrand,
   resolveRipeCountry,
 } from "./traffic-flow-brands.js"
@@ -44,6 +45,11 @@ assert.equal(isNamedInternetService("DNS", "DNS"), false)
 assert.equal(mapServiceNodeId("AWS"), "svc:aws")
 assert.equal(mapServiceNodeId("Cloudflare"), "svc:cloudflare")
 assert.equal(mapServiceNodeId("Прочее"), "svc:other")
+assert.equal(mapCountryNodeId("US"), "cc:us")
+assert.equal(mapCountryNodeId("nl"), "cc:nl")
+assert.equal(mapCountryNodeId(""), "cc:other")
+assert.equal(mapCountryNodeId("Прочее"), "cc:other")
+assert.equal(mapCountryNodeId("EU"), "cc:other")
 
 assert.equal(brandByAsn(714)?.service, "Apple")
 assert.equal(brandByAsn(714)?.category, "CDN")

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

@@ -364,3 +364,10 @@ export function mapServiceNodeId(label: string): string {
     .replace(/^-+|-+$/g, "")
   return `svc:${slug || "unknown"}`
 }
+
+/** `US` → `cc:us`; неизвестная / пустая → `cc:other`. */
+export function mapCountryNodeId(code: string): string {
+  const iso = normalizeIsoCountry(code)
+  if (!iso) return "cc:other"
+  return `cc:${iso.toLowerCase()}`
+}

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

@@ -306,6 +306,15 @@ try {
   assert.ok(googleEdge)
   assert.equal(googleEdge.clientName, "Alice")
   assert.equal((six.serviceEdges ?? []).reduce((n, e) => n + e.bytes, 0), 10_000)
+  const us = six.countries?.find((s) => s.id === "cc:us")
+  assert.ok(us, "Google ripe country US")
+  assert.equal(us.label, "US")
+  const usEdge = six.countryEdges?.find((e) => e.toId === "cc:us" && e.fromId === "9")
+  assert.ok(usEdge)
+  assert.ok(!(six.countryEdges ?? []).some((e) => e.toId.startsWith("svc:")), "страны не смешиваются с svc:*")
+  const usPath = six.countryPaths?.find((p) => p.serviceId === "cc:us" && p.enId === "9")
+  assert.ok(usPath)
+  assert.equal(usPath.clientName, "Alice")
 } finally {
   resetFlowRingsForTests()
   resetIfaceCacheForTests()
@@ -810,4 +819,51 @@ try {
   resetFlowCatalogForTests()
 }
 
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+seedFlowTopologyForTests(topo)
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "gre-client" },
+  { ".id": "*3", name: "gre-jh-en" },
+])
+googleRipe()
+seedRipeCacheForTests({
+  prefix: "185.45.12.0/24",
+  asn: 13335,
+  country: "NL",
+  lat: 52.3,
+  lng: 4.9,
+  holder: "CLOUDFLARENET, NL",
+  ok: true,
+  fetchedAt: Date.now(),
+})
+ingestParsedFlowsForServerForTests(7, [
+  payloadFlow("8.8.8.8", 5000),
+  payloadFlow("185.45.12.10", 5000),
+])
+try {
+  resetFlowMapHopsCacheForTests()
+  const split = await buildFlowMapHops({ minutes: 5, minSharePct: 5 })
+  const us = split.countries?.find((s) => s.id === "cc:us")
+  const nl = split.countries?.find((s) => s.id === "cc:nl")
+  assert.ok(us, "US из ripe Google")
+  assert.ok(nl, "NL из ripe Cloudflare")
+  assert.equal(us.bytes, 5000)
+  assert.equal(nl.bytes, 5000)
+  assert.ok(split.countryEdges?.some((e) => e.toId === "cc:us" && e.fromId === "9"))
+  assert.ok(split.countryEdges?.some((e) => e.toId === "cc:nl" && e.fromId === "9"))
+  assert.ok(!(split.countryEdges ?? []).some((e) => e.toId.startsWith("svc:")))
+  assert.ok(split.serviceEdges?.some((e) => e.toId === "svc:google"))
+  assert.ok(!(split.serviceEdges ?? []).some((e) => e.toId.startsWith("cc:")))
+  assert.ok(split.countryPaths?.some((p) => p.serviceId === "cc:nl" && p.enId === "9"))
+} finally {
+  seedFlowTopologyForTests(null)
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
 console.log("traffic-flow-map-hops.test.ts: ok")

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

@@ -3,7 +3,8 @@ import type { FlowMapHop, FlowMapHopsDto, FlowMapService, FlowMapServiceEdge, Fl
 import { db } from "../db/index.js"
 import { userInterfaceBindings } from "../db/schema.js"
 import { flowRowMatchesFilter } from "./traffic-flow-apps.js"
-import { OTHER_SERVICE, isNamedInternetService, mapServiceNodeId } from "./traffic-flow-brands.js"
+import { OTHER_SERVICE, isNamedInternetService, mapCountryNodeId, mapServiceNodeId, resolveRipeCountry } from "./traffic-flow-brands.js"
+import type { FlowIpMeta } from "./traffic-flow-ripe.js"
 import { refreshFlowCatalogInBackground } from "./traffic-flow-classify.js"
 import { dedupFlowRowsAcrossExporters, dedupFlowRowsMaxBytes } from "./traffic-flow-dedup.js"
 import { getFlowListenerState, listFlowRowsForWindow } from "./traffic-flow-ingest.js"
@@ -20,6 +21,7 @@ export const DEFAULT_MAP_SERVICE_MIN_SHARE_PCT = 5
 export const MAP_SERVICE_NODE_CAP = 20
 /** Минимум узлов-брендов на карте, даже если доля ниже порога. */
 export const MAP_SERVICE_MIN_NODES = 8
+export const MAP_COUNTRY_CATEGORY = "Страна"
 const HOPS_CACHE_TTL_MS = 2000
 
 export interface FlowMapHopsQuery {
@@ -64,6 +66,32 @@ interface DstAcc {
   fromBytes: Map<string, FromAcc>
 }
 
+interface DestTotal {
+  label: string
+  category: string
+  bytes: number
+}
+
+interface DestEdgeAcc {
+  fromId: string
+  toId: string
+  bytes: number
+  bytesFwd: number
+  bytesRev: number
+  clients: Map<string, string>
+}
+
+interface DestPathAcc {
+  clientId: string
+  clientName: string
+  viaId: string
+  viaName: string
+  enId: string
+  enName: string
+  serviceId: string
+  bytes: number
+}
+
 function bumpClient(clients: Map<string, ClientAcc>, bytes: number, client: { userId: string; name: string } | null): void {
   const id = client?.userId || "—"
   const name = client?.name || "—"
@@ -108,6 +136,137 @@ export function pickMapServices(ranked: FlowMapService[], minSharePct: number):
     .slice(0, MAP_SERVICE_NODE_CAP)
 }
 
+function bumpDestTotal(totals: Map<string, DestTotal>, id: string, label: string, category: string, bytes: number): void {
+  const prev = totals.get(id)
+  if (prev) {
+    prev.bytes += bytes
+    return
+  }
+  totals.set(id, { label, category, bytes })
+}
+
+function bumpDestFrom(
+  edges: Map<string, DestEdgeAcc>,
+  paths: Map<string, DestPathAcc>,
+  toId: string,
+  from: FromAcc,
+  exporterId: string,
+  fromId: string,
+  enName: string,
+  viaName: string,
+): void {
+  const edgeKey = `${fromId}|${toId}`
+  const prevEdge = edges.get(edgeKey)
+  const namedClients = new Map<string, string>()
+  for (const [id, c] of from.clients) {
+    if (id !== "—") namedClients.set(id, c.name)
+  }
+  if (prevEdge) {
+    prevEdge.bytes += from.bytes
+    prevEdge.bytesFwd += from.bytes
+    for (const [id, name] of namedClients) prevEdge.clients.set(id, name)
+  } else {
+    edges.set(edgeKey, {
+      fromId,
+      toId,
+      bytes: from.bytes,
+      bytesFwd: from.bytes,
+      bytesRev: 0,
+      clients: namedClients,
+    })
+  }
+  for (const [clientId, c] of from.clients) {
+    const pathKey = `${clientId}|${fromId}|${toId}`
+    const prevPath = paths.get(pathKey)
+    if (prevPath) {
+      prevPath.bytes += c.bytes
+      if (exporterId !== fromId && prevPath.viaId === fromId) {
+        prevPath.viaId = exporterId
+        prevPath.viaName = viaName
+      }
+      if (prevPath.clientName === "—" && c.name !== "—") prevPath.clientName = c.name
+    } else {
+      paths.set(pathKey, {
+        clientId,
+        clientName: c.name,
+        viaId: exporterId,
+        viaName,
+        enId: fromId,
+        enName,
+        serviceId: toId,
+        bytes: c.bytes,
+      })
+    }
+  }
+}
+
+function finalizeDestLayer(
+  totals: Map<string, DestTotal>,
+  edges: Map<string, DestEdgeAcc>,
+  paths: Map<string, DestPathAcc>,
+  windowSec: number,
+  minSharePct: number,
+  shareBase: number,
+): { nodes: FlowMapService[]; edges: FlowMapServiceEdge[]; paths: FlowMapServicePath[] } {
+  const nodes = pickMapServices(
+    [...totals.entries()]
+      .map(([id, s]) => ({
+        id,
+        label: s.label,
+        category: s.category,
+        bytes: s.bytes,
+        bps: (s.bytes * 8) / windowSec,
+        share: shareBase > 0 ? s.bytes / shareBase : 0,
+      }))
+      .sort((a, b) => b.bytes - a.bytes),
+    minSharePct,
+  )
+  const keep = new Set(nodes.map((s) => s.id))
+  const outEdges: FlowMapServiceEdge[] = [...edges.values()]
+    .filter((e) => keep.has(e.toId))
+    .map((e) => {
+      const clients = [...e.clients.entries()].map(([id, name]) => ({ id, name }))
+      const first = clients[0]
+      return {
+        fromId: e.fromId,
+        toId: e.toId,
+        bytes: e.bytes,
+        bps: (e.bytes * 8) / windowSec,
+        bpsFwd: (e.bytesFwd * 8) / windowSec,
+        bpsRev: (e.bytesRev * 8) / windowSec,
+        ...(first ? { clientId: first.id, clientName: first.name } : {}),
+        ...(clients.length ? { clients } : {}),
+      }
+    })
+    .sort((a, b) => b.bytes - a.bytes)
+  const outPaths: FlowMapServicePath[] = [...paths.values()]
+    .filter((p) => keep.has(p.serviceId))
+    .map((p) => ({
+      clientId: p.clientId,
+      clientName: p.clientName,
+      viaId: p.viaId,
+      viaName: p.viaName,
+      enId: p.enId,
+      enName: p.enName,
+      serviceId: p.serviceId,
+      bytes: p.bytes,
+      bps: (p.bytes * 8) / windowSec,
+    }))
+    .sort((a, b) => b.bps - a.bps)
+  return { nodes, edges: outEdges, paths: outPaths }
+}
+
+function countryDestFromRipe(ripe: FlowIpMeta | null, destKey: string): { id: string; label: string; category: string } {
+  if (!destKey || destKey === "__other__" || !ripe?.ok) {
+    return { id: mapCountryNodeId(""), label: OTHER_SERVICE, category: MAP_COUNTRY_CATEGORY }
+  }
+  const iso = resolveRipeCountry(ripe.country, ripe.asn, ripe.holder)
+  if (!iso) {
+    return { id: mapCountryNodeId(""), label: OTHER_SERVICE, category: MAP_COUNTRY_CATEGORY }
+  }
+  return { id: mapCountryNodeId(iso), label: iso, category: MAP_COUNTRY_CATEGORY }
+}
+
 function hopsQueryKey(q: FlowMapHopsQuery, minSharePct: number): string {
   return JSON.stringify({
     epoch: flowDataEpoch(),
@@ -368,25 +527,12 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
     }
   }
 
-  const svcTotals = new Map<string, { label: string; category: string; bytes: number }>()
-  const svcEdges = new Map<string, {
-    fromId: string
-    toId: string
-    bytes: number
-    bytesFwd: number
-    bytesRev: number
-    clients: Map<string, string>
-  }>()
-  const svcPaths = new Map<string, {
-    clientId: string
-    clientName: string
-    viaId: string
-    viaName: string
-    enId: string
-    enName: string
-    serviceId: string
-    bytes: number
-  }>()
+  const svcTotals = new Map<string, DestTotal>()
+  const svcEdges = new Map<string, DestEdgeAcc>()
+  const svcPaths = new Map<string, DestPathAcc>()
+  const ccTotals = new Map<string, DestTotal>()
+  const ccEdges = new Map<string, DestEdgeAcc>()
+  const ccPaths = new Map<string, DestPathAcc>()
 
   for (const h of hops.values()) {
     if (h.kind !== "gre" || !h.toId) continue
@@ -423,58 +569,17 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
     const classified = dst && dst !== "__other__"
       ? mapInternetBrand(dst, acc.proto, acc.dstPort, acc.srcPort, ripe)
       : { service: OTHER_SERVICE, category: OTHER_SERVICE }
-    const toId = mapServiceNodeId(classified.service)
-    const prevSvc = svcTotals.get(toId)
-    if (prevSvc) prevSvc.bytes += acc.bytes
-    else svcTotals.set(toId, { label: classified.service, category: classified.category, bytes: acc.bytes })
+    const svcId = mapServiceNodeId(classified.service)
+    bumpDestTotal(svcTotals, svcId, classified.service, classified.category, acc.bytes)
+    const country = countryDestFromRipe(ripe, dst)
+    bumpDestTotal(ccTotals, country.id, country.label, country.category, acc.bytes)
     for (const [exporterId, from] of acc.fromBytes) {
       const fromId = anchorEnId(exporterId)
       if (!fromId) continue
-      const edgeKey = `${fromId}|${toId}`
-      const prevEdge = svcEdges.get(edgeKey)
-      const namedClients = new Map<string, string>()
-      for (const [id, c] of from.clients) {
-        if (id !== "—") namedClients.set(id, c.name)
-      }
-      if (prevEdge) {
-        prevEdge.bytes += from.bytes
-        prevEdge.bytesFwd += from.bytes
-        for (const [id, name] of namedClients) prevEdge.clients.set(id, name)
-      } else {
-        svcEdges.set(edgeKey, {
-          fromId,
-          toId,
-          bytes: from.bytes,
-          bytesFwd: from.bytes,
-          bytesRev: 0,
-          clients: namedClients,
-        })
-      }
       const enName = nodeName(fromId)
       const viaName = nodeName(exporterId)
-      for (const [clientId, c] of from.clients) {
-        const pathKey = `${clientId}|${fromId}|${toId}`
-        const prevPath = svcPaths.get(pathKey)
-        if (prevPath) {
-          prevPath.bytes += c.bytes
-          if (exporterId !== fromId && prevPath.viaId === fromId) {
-            prevPath.viaId = exporterId
-            prevPath.viaName = viaName
-          }
-          if (prevPath.clientName === "—" && c.name !== "—") prevPath.clientName = c.name
-        } else {
-          svcPaths.set(pathKey, {
-            clientId,
-            clientName: c.name,
-            viaId: exporterId,
-            viaName,
-            enId: fromId,
-            enName,
-            serviceId: toId,
-            bytes: c.bytes,
-          })
-        }
-      }
+      bumpDestFrom(svcEdges, svcPaths, svcId, from, exporterId, fromId, enName, viaName)
+      bumpDestFrom(ccEdges, ccPaths, country.id, from, exporterId, fromId, enName, viaName)
     }
   }
 
@@ -483,52 +588,8 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
     .reduce((n, s) => n + s.bytes, 0)
   const unclassifiedBytes = Math.max(0, totalBytes - namedBytes)
   const shareBase = totalBytes > 0 ? totalBytes : namedBytes
-  const services = pickMapServices(
-    [...svcTotals.entries()]
-      .map(([id, s]) => ({
-        id,
-        label: s.label,
-        category: s.category,
-        bytes: s.bytes,
-        bps: (s.bytes * 8) / windowSec,
-        share: shareBase > 0 ? s.bytes / shareBase : 0,
-      }))
-      .sort((a, b) => b.bytes - a.bytes),
-    minSharePct,
-  )
-  const keepSvc = new Set(services.map((s) => s.id))
-  const serviceEdges: FlowMapServiceEdge[] = [...svcEdges.values()]
-    .filter((e) => keepSvc.has(e.toId))
-    .map((e) => {
-      const clients = [...e.clients.entries()].map(([id, name]) => ({ id, name }))
-      const first = clients[0]
-      return {
-        fromId: e.fromId,
-        toId: e.toId,
-        bytes: e.bytes,
-        bps: (e.bytes * 8) / windowSec,
-        bpsFwd: (e.bytesFwd * 8) / windowSec,
-        bpsRev: (e.bytesRev * 8) / windowSec,
-        ...(first ? { clientId: first.id, clientName: first.name } : {}),
-        ...(clients.length ? { clients } : {}),
-      }
-    })
-    .sort((a, b) => b.bytes - a.bytes)
-
-  const servicePaths: FlowMapServicePath[] = [...svcPaths.values()]
-    .filter((p) => keepSvc.has(p.serviceId))
-    .map((p) => ({
-      clientId: p.clientId,
-      clientName: p.clientName,
-      viaId: p.viaId,
-      viaName: p.viaName,
-      enId: p.enId,
-      enName: p.enName,
-      serviceId: p.serviceId,
-      bytes: p.bytes,
-      bps: (p.bytes * 8) / windowSec,
-    }))
-    .sort((a, b) => b.bps - a.bps)
+  const servicesOut = finalizeDestLayer(svcTotals, svcEdges, svcPaths, windowSec, minSharePct, shareBase)
+  const countriesOut = finalizeDestLayer(ccTotals, ccEdges, ccPaths, windowSec, minSharePct, shareBase)
 
   const listener = getFlowListenerState()
   const geo = geoipReadersStatus()
@@ -544,9 +605,12 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
     unclassifiedBytes,
     asnLoaded: geo.asnLoaded,
     countryLoaded: geo.countryLoaded,
-    services,
-    serviceEdges,
-    servicePaths,
+    services: servicesOut.nodes,
+    serviceEdges: servicesOut.edges,
+    servicePaths: servicesOut.paths,
+    countries: countriesOut.nodes,
+    countryEdges: countriesOut.edges,
+    countryPaths: countriesOut.paths,
     mapServiceMinSharePct: minSharePct,
     dedupApplied: wantDedup,
     excludeMeshApplied: excludeMesh,