feat(traffic-flow): enhance traffic classification and map display

- Added a new state variable `mapAsnLoaded` to track ASN loading status in the NetworkMapPage. - Updated the page to conditionally display a message when ASN data is not loaded, improving user feedback. - Enhanced traffic classification tests to include additional scenarios for Google and YouTube services, ensuring comprehensive coverage. - Introduced a new function `pickMapInternetDest` to improve destination selection logic in traffic flow processing. - Updated the traffic flow destination classification to handle new conditions and improve accuracy. Co-authored-by: Cursor <[email protected]>
feat(traffic-flow): enhance traffic flow analytics and overlay configuration
2026-09-12 01:07:26 +07:00 · 2026-09-12 00:03:13 +07:00 · 2026-09-11 23:01:48 +07:00
33 changed files with 1077 additions and 149 deletions
@@ -761,7 +761,7 @@ function ServiceNode({
      onMouseDown={(e) => { e.stopPropagation(); onMouseDown(e) }}
      onClick={(e) => { e.stopPropagation(); onClick() }}
    >
-      <title>{`${label} · ${serviceSharePct(share)} трафика окна`}</title>
+      <title>{`${label} · ${serviceSharePct(share)} payload окна`}</title>
      {isSel && (
        <rect
          x={-bw / 2 - 6}
@@ -1100,6 +1100,10 @@ export default function NetworkMapPage() {
  const [mapServiceEdges, setMapServiceEdges] = useState<FlowMapServiceEdge[]>([])
  const [mapServicePaths, setMapServicePaths] = useState<FlowMapServicePath[]>([])
  const [mapSharePct, setMapSharePct] = useState(5)
+  const [mapNamedBytes, setMapNamedBytes] = useState(0)
+  const [mapTotalBytes, setMapTotalBytes] = useState(0)
+  const [mapWindowSec, setMapWindowSec] = useState(300)
+  const [mapAsnLoaded, setMapAsnLoaded] = useState(true)
  /** FQDN из GRE outer → IPv4 (ответ POST /api/network/resolve-hosts), для матчинга с WAN. */
  const [greResolvedIpv4ByHost, setGreResolvedIpv4ByHost] = useState<Record<string, string>>({})
  const [dataError, setDataError] = useState<string | null>(null)
@@ -1196,6 +1200,8 @@ export default function NetworkMapPage() {
        setMapServiceEdges(MOCK_MAP_SERVICE_EDGES)
        setMapServicePaths(MOCK_MAP_SERVICE_PATHS)
        setMapSharePct(5)
+        setMapNamedBytes(0)
+        setMapTotalBytes(0)
        setDataError(null)
      })
      return
@@ -1303,6 +1309,10 @@ export default function NetworkMapPage() {
          setMapServiceEdges(res.serviceEdges ?? [])
          setMapServicePaths(res.servicePaths ?? [])
          if (res.mapServiceMinSharePct != null) setMapSharePct(res.mapServiceMinSharePct)
+          setMapNamedBytes(res.namedBytes ?? 0)
+          setMapTotalBytes(res.totalBytes ?? 0)
+          if (res.asnLoaded != null) setMapAsnLoaded(res.asnLoaded)
+          if (res.windowSec) setMapWindowSec(res.windowSec)
        })
        .catch((err: unknown) => {
          if (cancelled) return
@@ -2739,6 +2749,32 @@ export default function NetworkMapPage() {
                      <span className="text-xs text-muted-foreground">Доля окна</span>
                      <span className="text-xs font-mono font-medium text-cyan-400">{serviceSharePct(liveSelectedService.share)}</span>
                    </div>
+                    {mapTotalBytes > 0 && (
+                      <div className="flex items-center justify-between py-2 border-b border-border/50">
+                        <span className="text-xs text-muted-foreground">Классифицировано</span>
+                        <span className="text-xs font-mono font-medium text-cyan-400">
+                          {formatNetflowRate({
+                            bytes: mapNamedBytes,
+                            bps: (mapNamedBytes * 8) / Math.max(1, mapWindowSec),
+                            bpsFwd: 0,
+                            bpsRev: 0,
+                          })}
+                          {" из "}
+                          {formatNetflowRate({
+                            bytes: mapTotalBytes,
+                            bps: (mapTotalBytes * 8) / Math.max(1, mapWindowSec),
+                            bpsFwd: 0,
+                            bpsRev: 0,
+                          })}
+                        </span>
+                      </div>
+                    )}
+                    {!mapAsnLoaded && (
+                      <div className="flex items-center justify-between py-2 border-b border-border/50">
+                        <span className="text-xs text-muted-foreground">GeoLite2 ASN</span>
+                        <span className="text-xs font-mono font-medium text-amber-500">не загружена</span>
+                      </div>
+                    )}
                    <div className="flex items-center justify-between py-2 border-b border-border/50">
                      <span className="text-xs text-muted-foreground">Скорость</span>
                      <span className="text-xs font-mono font-medium">
@@ -0,0 +1,7 @@
+-- IPFIX postNAT (IANA 225/226) + postNAPT ports (IANA 227/228) from MikroTik Traffic Flow.
+-- Needed to rebuild facts with the same internet dest as the network map.
+
+ALTER TABLE flow_buckets ADD COLUMN IF NOT EXISTS nat_src INET;
+ALTER TABLE flow_buckets ADD COLUMN IF NOT EXISTS nat_dst INET;
+ALTER TABLE flow_buckets ADD COLUMN IF NOT EXISTS nat_src_port INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE flow_buckets ADD COLUMN IF NOT EXISTS nat_dst_port INTEGER NOT NULL DEFAULT 0;
@@ -9,6 +9,8 @@ if (!(await withPgOrSkip())) {
  process.exit(0)
 }

+await applySqlMigrations(pool)
+
 {
  const { rows } = await dbQuery<{ n: string }>(`SELECT COUNT(*)::text AS n FROM servers`)
  assert.ok(rows[0])
@@ -114,13 +116,17 @@ if (!(await withPgOrSkip())) {
    SELECT column_name, udt_name
    FROM information_schema.columns
    WHERE table_schema = 'public' AND table_name = 'flow_buckets'
-      AND column_name IN ('src', 'dst', 'next_hop', 'proto')
+      AND column_name IN ('src', 'dst', 'next_hop', 'proto', 'nat_src', 'nat_dst', 'nat_src_port', 'nat_dst_port')
  `)
  const by = Object.fromEntries(rows.map((r) => [r.column_name, r.udt_name]))
  assert.equal(by.src, "inet")
  assert.equal(by.dst, "inet")
  assert.equal(by.next_hop, "inet")
  assert.equal(by.proto, "int2")
+  assert.equal(by.nat_src, "inet")
+  assert.equal(by.nat_dst, "inet")
+  assert.equal(by.nat_src_port, "int4")
+  assert.equal(by.nat_dst_port, "int4")
 }

 {
@@ -179,6 +185,13 @@ if (!(await withPgOrSkip())) {
  assert.equal(by.section, "text")
 }

+{
+  const mig = await dbQuery<{ id: string }>(
+    `SELECT id FROM schema_migrations WHERE id = '0007_flow_buckets_nat'`,
+  )
+  assert.equal(mig.rows.length, 1, "0007 применена")
+}
+
 {
  const marker = await dbQuery<{ sqlite_imported_at: string | null }>(
    `SELECT sqlite_imported_at FROM data_migration WHERE id = 1`,
@@ -266,6 +266,10 @@ export const flowBuckets = pgTable("flow_buckets", {
  nextHop:   inet("next_hop"),
  flowStartMs: bigint("flow_start_ms", { mode: "number" }).notNull().default(0),
  flowEndMs:   bigint("flow_end_ms", { mode: "number" }).notNull().default(0),
+  natSrc:    inet("nat_src"),
+  natDst:    inet("nat_dst"),
+  natSrcPort: integer("nat_src_port").notNull().default(0),
+  natDstPort: integer("nat_dst_port").notNull().default(0),
 }, (t) => [
  primaryKey({
    name: "flow_buckets_pkey",
@@ -125,6 +125,7 @@ const TABLES: TableCopy[] = [
    ["src_port", "int"], ["dst_port", "int"], ["bytes", "int"], ["packets", "int"],
    ["in_iface", "text"], ["out_iface", "text"], ["next_hop", "inet"],
    ["flow_start_ms", "int"], ["flow_end_ms", "int"],
+    ["nat_src", "inet"], ["nat_dst", "inet"], ["nat_src_port", "int"], ["nat_dst_port", "int"],
  ]},
  { table: "flow_minute_stats", timeCol: "bucket_at", retentionDays: 3, columns: [
    ["server_id", "int"], ["bucket_at", "ts"], ["bytes", "int"], ["packets", "int"],
@@ -113,6 +113,7 @@ async function applyOverlayHandler(req: FastifyRequest, reply: FastifyReply) {
    const result = await applyFlowOverlay(parsed.data.serverId, {
      publicEndpoint: parsed.data.publicEndpoint,
      requestHost: requestPublicHost(req),
+      disableGreFastPath: parsed.data.disableGreFastPath,
    })
    return reply.send(result)
  } catch (e) {
@@ -388,8 +388,8 @@ try {
    assert.equal(def.excludeOverlayApplied, true)
    assert.equal(def.excludeMeshApplied, true)
    assert.ok(!def.conversationsList.some((r) => r.proto === 47))
-    assert.equal(def.conversationsList[0]?.service, "Google")
-    assert.equal(def.conversationsList[0]?.category, "Веб")
+    assert.equal(def.conversationsList[0]?.service, "YouTube")
+    assert.equal(def.conversationsList[0]?.category, "Видео / стриминг")
    assert.equal(def.conversationsList[0]?.clientName, "Alice")
    assert.equal(def.conversationsList[0]?.enName, "NSK-SERVHOST-RTK")
    assert.equal(def.conversationsList[0]?.plane, "payload")
@@ -445,8 +445,8 @@ try {
    const rev = await buildFlowAnalytics({ minutes: 5, serverId: 7 })
    const google = rev.conversationsList.find((r) => r.src === "173.194.151.65")
    const cf = rev.conversationsList.find((r) => r.src === "104.18.35.51")
-    assert.equal(google?.service, "Google")
-    assert.equal(google?.category, "Веб")
+    assert.equal(google?.service, "YouTube")
+    assert.equal(google?.category, "Видео / стриминг")
    assert.equal(cf?.service, "Cloudflare")
    assert.equal(cf?.category, "CDN")
  } finally {
@@ -260,6 +260,10 @@ async function buildFlowAnalyticsUncached(q: FlowAnalyticsQuery): Promise<FlowAn
      serverId: r.serverId,
      inIface: resolved.name,
      topo,
+      natSrc: r.natSrc,
+      natDst: r.natDst,
+      natSrcPort: r.natSrcPort,
+      natDstPort: r.natDstPort,
    })
    if (destMeta.dest) peers.add(destMeta.dest)
    const app = applicationName(r.proto, r.dstPort, r.srcPort)
@@ -32,6 +32,8 @@ assert.equal(brandByAsn(401115)?.service, "ChatGPT")
 assert.equal(lookupBrand("1.1.1.1", 13335)?.service, "Cloudflare")
 assert.equal(lookupBrand("104.18.35.51", 0)?.service, "Cloudflare")
 assert.equal(lookupBrand("173.194.151.65", 0)?.service, "Google")
+assert.equal(lookupBrand("64.233.161.1", 0)?.service, "Google")
+assert.equal(lookupBrand("142.250.1.10", 0)?.service, "Google")
 assert.equal(lookupBrand("8.8.8.8", 0)?.service, "Google")
 assert.equal(lookupBrand("203.0.113.9", 64500), null)
 assert.equal(OTHER_SERVICE, "Прочее")
@@ -41,6 +43,7 @@ assert.equal(isNamedInternetService("GRE", "Туннель"), false)
 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(brandByAsn(714)?.service, "Apple")
 assert.equal(brandByAsn(714)?.category, "CDN")
@@ -70,6 +73,12 @@ assert.equal(isSteamGamePort(6, 443, 50000), false)

 assert.equal(resolveFlowBrand("104.18.35.51", 32590, "VALVE-CORPORATION", 6, 443, 1)?.service, "Cloudflare")
 assert.equal(resolveFlowBrand("203.0.113.9", 32590, "", 17, 27015, 50000)?.service, "Steam")
+assert.equal(resolveFlowBrand("8.8.8.8", 15169, "GOOGLE", 6, 443, 51234)?.service, "Google")
+assert.equal(resolveFlowBrand("173.194.160.163", 15169, "GOOGLE", 6, 443, 51234)?.service, "YouTube")
+assert.equal(resolveFlowBrand("64.233.161.1", 0, "", 17, 443, 50000)?.service, "YouTube")
+assert.equal(resolveFlowBrand("64.233.161.1", 0, "", 6, 80, 50000)?.service, "Google")
+assert.equal(resolveFlowBrand("2001:4860:4860::8888", 15169, "GOOGLE", 17, 53, 53000)?.service, "Google")
+assert.equal(resolveFlowBrand("2001:4860:4860::8888", 15169, "GOOGLE", 17, 443, 50000)?.service, "YouTube")
 assert.equal(resolveRipeCountry("", 9059, ""), "IE")
 assert.equal(resolveRipeCountry("", 24940, ""), "DE")

@@ -174,6 +174,14 @@ const CIDR_BRANDS: Array<{ cidr: string; prefixLen: number; hit: BrandHit }> = [
  { cidr: "172.217.0.0/16", prefixLen: 16, hit: GOOGLE },
  { cidr: "74.125.0.0/16", prefixLen: 16, hit: GOOGLE },
  { cidr: "142.250.0.0/15", prefixLen: 15, hit: GOOGLE },
+  { cidr: "64.233.0.0/16", prefixLen: 16, hit: GOOGLE },
+  { cidr: "66.102.0.0/16", prefixLen: 16, hit: GOOGLE },
+  { cidr: "66.249.64.0/19", prefixLen: 19, hit: GOOGLE },
+  { cidr: "72.14.192.0/18", prefixLen: 18, hit: GOOGLE },
+  { cidr: "108.177.0.0/16", prefixLen: 16, hit: GOOGLE },
+  { cidr: "209.85.128.0/17", prefixLen: 17, hit: GOOGLE },
+  { cidr: "216.58.192.0/19", prefixLen: 19, hit: GOOGLE },
+  { cidr: "216.239.32.0/19", prefixLen: 19, hit: GOOGLE },
  { cidr: "208.65.152.0/22", prefixLen: 22, hit: YOUTUBE },
  { cidr: "208.117.224.0/19", prefixLen: 19, hit: YOUTUBE },
 ].sort((a, b) => b.prefixLen - a.prefixLen)
@@ -196,6 +204,16 @@ const HOLDER_BRANDS: Array<{ re: RegExp; hit: BrandHit }> = [
 const NON_ISO = new Set(["EU", "AP", "ZZ", "XX", "A1", "A2", "O1"])

 const STEAM_ASN = 32590
+const GOOGLE_FRONT_ASN = new Set([15169, 396982])
+
+function isGooglePublicDns(ip: string): boolean {
+  return ipInCidrV4(ip, "8.8.8.0/24") || ipInCidrV4(ip, "8.8.4.0/24")
+}
+
+function isHttpsOrQuic(proto: number, dstPort: number, srcPort: number): boolean {
+  if (proto !== 6 && proto !== 17) return false
+  return dstPort === 443 || srcPort === 443
+}

 export function isIsoCountry(code: string): boolean {
  const c = String(code ?? "").trim().toUpperCase()
@@ -273,7 +291,15 @@ export function resolveFlowBrand(
  if (cidrBrand?.service === "Cloudflare") return cidrBrand
  const holderBrand = brandByHolder(holder)
  if (holderBrand) return holderBrand
-  const fromLookup = cidrBrand || brandByAsn(asn)
+  const asnBrand = brandByAsn(asn)
+  if (
+    !isGooglePublicDns(ip)
+    && isHttpsOrQuic(proto, dstPort, srcPort)
+    && (GOOGLE_FRONT_ASN.has(asn) || cidrBrand?.service === "Google" || asnBrand?.service === "Google")
+  ) {
+    return YOUTUBE
+  }
+  const fromLookup = cidrBrand || asnBrand
  if (fromLookup) return fromLookup
  if (asn === STEAM_ASN && isSteamGamePort(proto, dstPort, srcPort)) return STEAM
  return null
@@ -300,8 +326,9 @@ export function isNamedInternetService(service: string, category: string): boole
 }

 export function mapServiceNodeId(label: string): string {
-  const slug = label
-    .trim()
+  const raw = label.trim()
+  if (raw === OTHER_SERVICE) return "svc:other"
+  const slug = raw
    .toLowerCase()
    .replace(/[^a-z0-9]+/g, "-")
    .replace(/^-+|-+$/g, "")
@@ -33,10 +33,10 @@ const google = classifyFlowDst("173.194.160.163", 6, 443, 1, {
  ok: true,
  fetchedAt: Date.now(),
 })
-assert.equal(google.service, "Google")
-assert.equal(google.category, "Веб")
+assert.equal(google.service, "YouTube")
+assert.equal(google.category, "Видео / стриминг")

-const googleCidr = classifyFlowDst("173.194.151.65", 6, 57182, 443, null)
+const googleCidr = classifyFlowDst("173.194.151.65", 6, 80, 50000, null)
 assert.equal(googleCidr.service, "Google")
 assert.equal(googleCidr.category, "Веб")

@@ -115,12 +115,46 @@ const googleCloud = classifyFlowDst("203.0.113.43", 6, 443, 1, {
  ok: true,
  fetchedAt: Date.now(),
 })
-assert.equal(googleCloud.service, "Google")
-assert.equal(googleCloud.category, "Веб")
+assert.equal(googleCloud.service, "YouTube")
+assert.equal(googleCloud.category, "Видео / стриминг")

 const gre = classifyFlowDst("198.51.100.1", 47, 0, 0, null)
 assert.equal(gre.service, "GRE")
 assert.equal(gre.category, "Туннель")
+const greIgnore = classifyFlowDst("8.8.8.8", 47, 0, 0, {
+  prefix: "8.8.8.0/24",
+  asn: 15169,
+  country: "US",
+  lat: null,
+  lng: null,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+}, { ignoreTunnelProto: true })
+assert.equal(greIgnore.service, "Google")
+const dnsGoogle = classifyFlowDst("8.8.8.8", 17, 53, 53000, {
+  prefix: "8.8.8.0/24",
+  asn: 15169,
+  country: "US",
+  lat: null,
+  lng: null,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+})
+assert.equal(dnsGoogle.service, "Google")
+assert.notEqual(dnsGoogle.service, "Прочее")
+const ipv6Yt = classifyFlowDst("2001:4860:4860::8888", 17, 443, 50000, {
+  prefix: "2001:4860:4860::8888/128",
+  asn: 15169,
+  country: "US",
+  lat: null,
+  lng: null,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+})
+assert.equal(ipv6Yt.service, "YouTube")
 const esp = classifyFlowDst("198.51.100.1", 50, 0, 0, null)
 assert.equal(esp.category, "Туннель")
 assert.equal(applicationName(17, 443, 50000), "QUIC")
@@ -73,9 +73,12 @@ export function classifyFlowDst(
  dstPort: number,
  srcPort: number,
  ripe: FlowIpMeta | null,
+  opts?: { ignoreTunnelProto?: boolean },
 ): FlowClassification {
-  if (proto === 47) return { service: "GRE", category: "Туннель" }
-  if (proto === 50) return { service: "ESP", category: "Туннель" }
+  if (!opts?.ignoreTunnelProto) {
+    if (proto === 47) return { service: "GRE", category: "Туннель" }
+    if (proto === 50) return { service: "ESP", category: "Туннель" }
+  }
  const app = applicationName(proto, dstPort, srcPort)
  if (app === "WireGuard") return { service: "WireGuard", category: "Туннель" }
  const hit = matchCidr(dst)
@@ -4,6 +4,7 @@ import {
  resetEngineForTests,
 } from "./traffic-flow-engine.js"
 import { factsSnapshotForTests } from "./traffic-flow-facts.js"
+import { classifyInternetBrand, mapInternetBrand } from "./traffic-flow-dest.js"
 import { disableCatalogFetchForTests, resetFlowCatalogForTests } from "./traffic-flow-classify.js"
 import {
  disableRipeEnqueueForTests,
@@ -88,19 +89,98 @@ ingestParsedFlowsForServerForTests(1, [
    inIface: "gre-client",
    outIface: "ether1",
  },
+  {
+    src: "203.0.113.10",
+    dst: "198.51.100.1",
+    proto: 47,
+    srcPort: 0,
+    dstPort: 0,
+    bytes: 9_000,
+    packets: 90,
+    inIface: "NSK-SERVHOST-RTK",
+    outIface: "NSK-SERVHOST-RTK",
+  },
+  {
+    src: "10.200.100.53",
+    dst: "10.200.100.1",
+    proto: 6,
+    srcPort: 53880,
+    dstPort: 443,
+    bytes: 70,
+    packets: 1,
+    inIface: "gre-client",
+    outIface: "ether1",
+    natDst: "8.8.8.8",
+    natDstPort: 443,
+  },
 ])

 const facts = factsSnapshotForTests()
 const total = facts.reduce((s, r) => s + r.bytes, 0)
-const asnBytes = facts.reduce((s, r) => s + r.bytes, 0)
-assert.equal(total, 150)
-assert.equal(asnBytes, 150, "unique bytes = SUM dest ASN")
+assert.equal(total, 120, "unique = Google payload + NAT, без overlay/пустого dest")
 assert.equal(facts.some((r) => r.asn === 12389), false, "ASN клиента не в кубе")
+assert.equal(facts.some((r) => r.service === "GRE"), false, "GRE не сервис unique")
 const google = facts.find((r) => r.asn === 15169)
 assert.ok(google)
-assert.equal(google.bytes, 50)
-const other = facts.filter((r) => r.asn === 0).reduce((s, r) => s + r.bytes, 0)
-assert.equal(other, 100)
+assert.equal(google.bytes, 120)
+assert.equal(facts.filter((r) => r.asn === 0).reduce((s, r) => s + r.bytes, 0), 0)
+
+assert.equal(classifyInternetBrand("8.8.8.8", 47, 0, 0, null), null, "GRE не бренд")
+assert.equal(classifyInternetBrand("8.8.8.8", 6, 443, 51234, {
+  prefix: "8.8.8.0/24",
+  asn: 15169,
+  country: "US",
+  lat: null,
+  lng: null,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+})?.service, "Google")
+assert.equal(mapInternetBrand("203.0.113.50", 6, 443, 51234, null).service, "Прочее")
+assert.equal(mapInternetBrand("8.8.8.8", 47, 0, 0, null).service, "Прочее")
+assert.equal(mapInternetBrand("8.8.8.8", 6, 443, 51234, {
+  prefix: "8.8.8.0/24",
+  asn: 15169,
+  country: "US",
+  lat: null,
+  lng: null,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+}).service, "Google")
+
+assert.equal(classifyInternetBrand("8.8.8.8", 17, 53, 53000, {
+  prefix: "8.8.8.0/24",
+  asn: 15169,
+  country: "US",
+  lat: null,
+  lng: null,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+})?.service, "Google")
+assert.equal(mapInternetBrand("8.8.8.8", 17, 53, 53000, {
+  prefix: "8.8.8.0/24",
+  asn: 15169,
+  country: "US",
+  lat: null,
+  lng: null,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+}).service, "Google")
+assert.equal(mapInternetBrand("2001:4860:4860::8888", 17, 443, 50000, {
+  prefix: "2001:4860:4860::8888/128",
+  asn: 15169,
+  country: "US",
+  lat: null,
+  lng: null,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+}).service, "YouTube")
+assert.equal(mapInternetBrand("64.233.161.1", 17, 443, 50000, null).service, "YouTube")
+assert.equal(mapInternetBrand("142.250.1.10", 6, 443, 1, null).service, "YouTube")

 resetEngineForTests()
 seedFlowTopologyForTests(null)
@@ -1,4 +1,5 @@
-import { isIsoCountry } from "./traffic-flow-brands.js"
+import { applicationName } from "./traffic-flow-apps.js"
+import { isIsoCountry, isNamedInternetService, OTHER_SERVICE, resolveFlowBrand } from "./traffic-flow-brands.js"
 import { classifyFlowDst, type FlowClassification } from "./traffic-flow-classify.js"
 import { resolveFlowIp } from "./traffic-flow-geoip.js"
 import { canonicalFactIface } from "./traffic-flow-ifindex.js"
@@ -22,14 +23,61 @@ export function destCtxForIface(
  topo: FlowTopology | null | undefined,
  serverId: number,
  inIface: string,
+  nat?: Pick<InternetDestCtx, "natSrc" | "natDst" | "natSrcPort" | "natDstPort">,
 ): InternetDestCtx {
  const name = canonicalFactIface(serverId, inIface) || String(inIface ?? "").trim()
  return {
    ours: flowOursHosts(topo),
-    boundClient: Boolean(topo && name && resolveClient(topo, serverId, name)),
+    boundClient: Boolean(
+      topo && name && (
+        resolveClient(topo, serverId, name)
+        || topo.clientIfaces.get(serverId)?.has(name)
+      ),
+    ),
+    natSrc: nat?.natSrc,
+    natDst: nat?.natDst,
+    natSrcPort: nat?.natSrcPort,
+    natDstPort: nat?.natDstPort,
  }
 }

+const OTHER_BRAND: FlowClassification = { service: OTHER_SERVICE, category: OTHER_SERVICE }
+
+function isTunnelProto(proto: number, dstPort: number, srcPort: number): boolean {
+  if (proto === 47 || proto === 50) return true
+  return applicationName(proto, dstPort, srcPort) === "WireGuard"
+}
+
+/** Бренд интернет-dest: ASN/CIDR до skip DNS. GRE/ESP/WG — не сервис. */
+export function classifyInternetBrand(
+  dst: string,
+  proto: number,
+  dstPort: number,
+  srcPort: number,
+  ripe: FlowIpMeta | null,
+): FlowClassification | null {
+  if (isTunnelProto(proto, dstPort, srcPort)) return null
+  const brand = resolveFlowBrand(dst, ripe?.asn ?? 0, ripe?.holder ?? "", proto, dstPort, srcPort)
+  if (brand && isNamedInternetService(brand.service, brand.category)) return brand
+  const app = applicationName(proto, dstPort, srcPort)
+  if (app === "DNS" || app === "SSH" || app === "BGP") return null
+  return null
+}
+
+/** Тот же классификатор, что аналитика (GeoLite2 ASN + catalog). Туннель → Прочее. */
+export function mapInternetBrand(
+  dst: string,
+  proto: number,
+  dstPort: number,
+  srcPort: number,
+  ripe: FlowIpMeta | null,
+): FlowClassification {
+  if (isTunnelProto(proto, dstPort, srcPort)) return OTHER_BRAND
+  const classified = classifyFlowDst(dst, proto, dstPort, srcPort, ripe, { ignoreTunnelProto: true })
+  if (isNamedInternetService(classified.service, classified.category)) return classified
+  return OTHER_BRAND
+}
+
 export function resolveInternetDest(opts: {
  src: string
  dst: string
@@ -39,16 +87,27 @@ export function resolveInternetDest(opts: {
  serverId: number
  inIface: string
  topo?: FlowTopology | null
+  natSrc?: string
+  natDst?: string
+  natSrcPort?: number
+  natDstPort?: number
 }): InternetDestMeta {
  const dest = pickInternetDest(
    opts.src,
    opts.dst,
    opts.srcPort,
    opts.dstPort,
-    destCtxForIface(opts.topo, opts.serverId, opts.inIface),
+    destCtxForIface(opts.topo, opts.serverId, opts.inIface, {
+      natSrc: opts.natSrc,
+      natDst: opts.natDst,
+      natSrcPort: opts.natSrcPort,
+      natDstPort: opts.natDstPort,
+    }),
  )
  const ripe = dest ? resolveFlowIp(dest) : null
-  const classified = classifyFlowDst(dest || opts.dst, opts.proto, opts.dstPort, opts.srcPort, ripe)
+  const classified = dest
+    ? classifyFlowDst(dest, opts.proto, opts.dstPort, opts.srcPort, ripe, { ignoreTunnelProto: true })
+    : classifyFlowDst(opts.dst, opts.proto, opts.dstPort, opts.srcPort, ripe)
  if (!dest) {
    return { dest: "", ripe: null, classified, country: "", asn: 0 }
  }
@@ -61,6 +61,10 @@ export interface PendingFlowRow {
  nextHop: string
  flowStartMs: number
  flowEndMs: number
+  natSrc: string
+  natDst: string
+  natSrcPort: number
+  natDstPort: number
 }

 function inetOrNull(value: string | null | undefined): string | null {
@@ -90,6 +94,17 @@ function clampProto(n: number): number {
  return Math.max(0, Math.min(255, Math.trunc(n)))
 }

+function clampPort(n: number): number {
+  if (!Number.isFinite(n)) return 0
+  return Math.max(0, Math.min(65535, Math.trunc(n)))
+}
+
+function sanitizeNatIp(value: string | null | undefined): string {
+  const s = String(value ?? "").trim()
+  if (!s || s === "0.0.0.0") return ""
+  return isValidFlowInet(s) ? s : ""
+}
+
 function sanitizeFlowRow(r: PendingFlowRow): PendingFlowRow | null {
  const src = (r.src || "").trim() || "0.0.0.0"
  const dst = (r.dst || "").trim() || "0.0.0.0"
@@ -101,6 +116,10 @@ function sanitizeFlowRow(r: PendingFlowRow): PendingFlowRow | null {
    dst,
    nextHop: next && isValidFlowInet(next) ? next : "",
    proto: clampProto(r.proto),
+    natSrc: sanitizeNatIp(r.natSrc),
+    natDst: sanitizeNatIp(r.natDst),
+    natSrcPort: clampPort(r.natSrcPort),
+    natDstPort: clampPort(r.natDstPort),
  }
 }

@@ -120,6 +139,10 @@ function flowUpsertParams(r: PendingFlowRow) {
    nextHop: inetOrNull(r.nextHop),
    flowStartMs: r.flowStartMs,
    flowEndMs: r.flowEndMs,
+    natSrc: inetOrNull(r.natSrc),
+    natDst: inetOrNull(r.natDst),
+    natSrcPort: r.natSrcPort,
+    natDstPort: r.natDstPort,
  }
 }

@@ -360,6 +383,10 @@ export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): vo
      serverId,
      inIface: flow.inIface,
      topo,
+      natSrc: flow.natSrc,
+      natDst: flow.natDst,
+      natSrcPort: flow.natSrcPort,
+      natDstPort: flow.natDstPort,
    })
    const ripe = destMeta.ripe
    if (destMeta.dest && !ripe) ripeMisses.push(destMeta.dest)
@@ -385,6 +412,11 @@ export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): vo
      src: flow.src,
      dst: flow.dst,
      topo,
+      dest: destMeta.dest,
+      natSrc: flow.natSrc,
+      natDst: flow.natDst,
+      natSrcPort: flow.natSrcPort,
+      natDstPort: flow.natDstPort,
    })) {
      bumpFlowFact({
        serverId,
@@ -405,6 +437,10 @@ export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): vo
      prev.packets += flow.packets
      if (flow.outIface && !prev.flow.outIface) prev.flow.outIface = flow.outIface
      if (flow.nextHop && !prev.flow.nextHop) prev.flow.nextHop = flow.nextHop
+      if (flow.natSrc && !prev.flow.natSrc) prev.flow.natSrc = flow.natSrc
+      if (flow.natDst && !prev.flow.natDst) prev.flow.natDst = flow.natDst
+      if (flow.natSrcPort && !prev.flow.natSrcPort) prev.flow.natSrcPort = flow.natSrcPort
+      if (flow.natDstPort && !prev.flow.natDstPort) prev.flow.natDstPort = flow.natDstPort
      if (flow.flowStartMs && (!prev.flow.flowStartMs || flow.flowStartMs < prev.flow.flowStartMs)) {
        prev.flow.flowStartMs = flow.flowStartMs
      }
@@ -460,6 +496,10 @@ function toPendingRow(row: PendingEntry): PendingFlowRow {
    nextHop: flow.nextHop,
    flowStartMs: flow.flowStartMs,
    flowEndMs: flow.flowEndMs,
+    natSrc: flow.natSrc,
+    natDst: flow.natDst,
+    natSrcPort: flow.natSrcPort,
+    natDstPort: flow.natDstPort,
  }
 }

@@ -471,6 +511,10 @@ function mergeInto(map: Map<string, PendingFlowRow>, row: PendingFlowRow): void
    prev.packets += row.packets
    if (row.outIface && !prev.outIface) prev.outIface = row.outIface
    if (row.nextHop && !prev.nextHop) prev.nextHop = row.nextHop
+    if (row.natSrc && !prev.natSrc) prev.natSrc = row.natSrc
+    if (row.natDst && !prev.natDst) prev.natDst = row.natDst
+    if (row.natSrcPort && !prev.natSrcPort) prev.natSrcPort = row.natSrcPort
+    if (row.natDstPort && !prev.natDstPort) prev.natDstPort = row.natDstPort
    if (row.flowStartMs && (!prev.flowStartMs || row.flowStartMs < prev.flowStartMs)) prev.flowStartMs = row.flowStartMs
    if (row.flowEndMs > (prev.flowEndMs ?? 0)) prev.flowEndMs = row.flowEndMs
    return
@@ -779,7 +823,7 @@ async function upsertFlowBucketsBatch(rows: PendingFlowRow[]): Promise<void> {
  await pool.query({
    text: `
      INSERT INTO flow_buckets (
-        server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms
+        server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms, nat_src, nat_dst, nat_src_port, nat_dst_port
      )
      SELECT *
      FROM UNNEST(
@@ -796,8 +840,12 @@ async function upsertFlowBucketsBatch(rows: PendingFlowRow[]): Promise<void> {
        $11::text[],
        $12::inet[],
        $13::bigint[],
-        $14::bigint[]
-      ) AS t(server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms)
+        $14::bigint[],
+        $15::inet[],
+        $16::inet[],
+        $17::int[],
+        $18::int[]
+      ) AS t(server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms, nat_src, nat_dst, nat_src_port, nat_dst_port)
      ON CONFLICT (server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
      DO UPDATE SET
        bytes = flow_buckets.bytes + excluded.bytes,
@@ -807,7 +855,11 @@ async function upsertFlowBucketsBatch(rows: PendingFlowRow[]): Promise<void> {
        flow_start_ms = CASE
          WHEN excluded.flow_start_ms > 0 AND (flow_buckets.flow_start_ms = 0 OR excluded.flow_start_ms < flow_buckets.flow_start_ms)
          THEN excluded.flow_start_ms ELSE flow_buckets.flow_start_ms END,
-        flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms)
+        flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms),
+        nat_src = COALESCE(excluded.nat_src, flow_buckets.nat_src),
+        nat_dst = COALESCE(excluded.nat_dst, flow_buckets.nat_dst),
+        nat_src_port = CASE WHEN excluded.nat_src_port > 0 THEN excluded.nat_src_port ELSE flow_buckets.nat_src_port END,
+        nat_dst_port = CASE WHEN excluded.nat_dst_port > 0 THEN excluded.nat_dst_port ELSE flow_buckets.nat_dst_port END
    `,
    values: [
      rows.map((r) => r.serverId),
@@ -824,15 +876,19 @@ async function upsertFlowBucketsBatch(rows: PendingFlowRow[]): Promise<void> {
      rows.map((r) => inetOrNull(r.nextHop)),
      rows.map((r) => r.flowStartMs),
      rows.map((r) => r.flowEndMs),
+      rows.map((r) => inetOrNull(r.natSrc)),
+      rows.map((r) => inetOrNull(r.natDst)),
+      rows.map((r) => r.natSrcPort),
+      rows.map((r) => r.natDstPort),
    ],
  })
 }

 const FLOW_UPSERT_SQL = `
    INSERT INTO flow_buckets (
-      server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms
+      server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms, nat_src, nat_dst, nat_src_port, nat_dst_port
    ) VALUES (
-      @serverId, @bucketAt, @src, @dst, @proto, @srcPort, @dstPort, @bytes, @packets, @inIface, @outIface, @nextHop, @flowStartMs, @flowEndMs
+      @serverId, @bucketAt, @src, @dst, @proto, @srcPort, @dstPort, @bytes, @packets, @inIface, @outIface, @nextHop, @flowStartMs, @flowEndMs, @natSrc, @natDst, @natSrcPort, @natDstPort
    )
    ON CONFLICT(server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
    DO UPDATE SET
@@ -843,7 +899,11 @@ const FLOW_UPSERT_SQL = `
      flow_start_ms = CASE
        WHEN excluded.flow_start_ms > 0 AND (flow_buckets.flow_start_ms = 0 OR excluded.flow_start_ms < flow_buckets.flow_start_ms)
        THEN excluded.flow_start_ms ELSE flow_buckets.flow_start_ms END,
-      flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms)
+      flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms),
+      nat_src = COALESCE(excluded.nat_src, flow_buckets.nat_src),
+      nat_dst = COALESCE(excluded.nat_dst, flow_buckets.nat_dst),
+      nat_src_port = CASE WHEN excluded.nat_src_port > 0 THEN excluded.nat_src_port ELSE flow_buckets.nat_src_port END,
+      nat_dst_port = CASE WHEN excluded.nat_dst_port > 0 THEN excluded.nat_dst_port ELSE flow_buckets.nat_dst_port END
  `

 async function upsertFlowBuckets(rows: PendingFlowRow[]): Promise<number> {
@@ -145,6 +145,63 @@ const enWan = shouldWriteFlowFact({
 })
 assert.equal(enWan, true, "WAN payload на EN — да")

+const emptyDest = shouldWriteFlowFact({
+  serverId: 1,
+  serverType: "jump-host",
+  inIface: "gre-client",
+  outIface: "ether1",
+  proto: 6,
+  srcPort: 51234,
+  dstPort: 443,
+  src: "95.167.1.10",
+  dst: "10.200.100.53",
+  topo: topo(),
+})
+assert.equal(emptyDest, false, "пустой интернет-dest не в facts")
+
+const jhToEnHosts = shouldWriteFlowFact({
+  serverId: 1,
+  serverType: "jump-host",
+  inIface: "ether1",
+  proto: 6,
+  srcPort: 0,
+  dstPort: 0,
+  src: "203.0.113.10",
+  dst: "198.51.100.1",
+  topo: topo(),
+})
+assert.equal(jhToEnHosts, false, "JH↔EN hosts не dest")
+
+const overlayNamed = shouldWriteFlowFact({
+  serverId: 1,
+  serverType: "jump-host",
+  inIface: "NSK-SERVHOST-RTK",
+  outIface: "NSK-SERVHOST-RTK",
+  proto: 47,
+  srcPort: 0,
+  dstPort: 0,
+  src: "203.0.113.10",
+  dst: "198.51.100.1",
+  topo: typed,
+})
+assert.equal(overlayNamed, false, "overlay proto 47 на NSK-SERVHOST-RTK не в facts")
+
+const natPayload = shouldWriteFlowFact({
+  serverId: 1,
+  serverType: "jump-host",
+  inIface: "gre-client",
+  outIface: "ether1",
+  proto: 6,
+  srcPort: 53880,
+  dstPort: 443,
+  src: "10.200.100.53",
+  dst: "10.200.100.1",
+  natDst: "8.8.8.8",
+  natDstPort: 443,
+  topo: topo(),
+})
+assert.equal(natPayload, true, "NAT Google на client GRE — да")
+
 seedFlowTopologyForTests(null)
 resetIfaceCacheForTests()
 console.log("traffic-flow-facts-filter.test.ts: ok")
@@ -1,8 +1,10 @@
 import { mapRosInterfaceType } from "../modules/users/iface-type.js"
 import { STATISTICS_DUP_MARK, STATISTICS_WAN_MARK } from "@mmapp/contracts/statistics"
+import { destCtxForIface } from "./traffic-flow-dest.js"
 import { canonicalFactIface } from "./traffic-flow-ifindex.js"
-import { classifyFlowPlane } from "./traffic-flow-planes.js"
-import { resolveClient, type FlowTopology } from "./traffic-flow-topology.js"
+import { pickInternetDest, isLocalIp } from "./traffic-flow-ip.js"
+import { classifyFlowPlane, isTunnelProto } from "./traffic-flow-planes.js"
+import { flowOursHosts, resolveClient, type FlowTopology } from "./traffic-flow-topology.js"

 const JUNK_IFACE = new Set(["", "0", "—", "__unknown__", "wg-flow"])

@@ -82,6 +84,11 @@ export function shouldWriteFlowFact(opts: {
  src: string
  dst: string
  topo?: FlowTopology | null
+  dest?: string
+  natSrc?: string
+  natDst?: string
+  natSrcPort?: number
+  natDstPort?: number
 }): boolean {
  const inName = canonicalFactIface(opts.serverId, opts.inIface) || String(opts.inIface ?? "").trim()
  if (isJunkFactIface(inName) || isJunkFactIface(opts.inIface)) return false
@@ -96,7 +103,25 @@ export function shouldWriteFlowFact(opts: {
    inIface: inName,
    outIface: outName || undefined,
  }, opts.topo?.plane)
-  if (plane !== "payload") return false
+  if (plane === "mgmt") return false
+  if (plane === "overlay" || isTunnelProto(opts.proto, opts.srcPort, opts.dstPort)) return false
+  const dest = opts.dest !== undefined
+    ? opts.dest
+    : pickInternetDest(
+      opts.src,
+      opts.dst,
+      opts.srcPort,
+      opts.dstPort,
+      destCtxForIface(opts.topo, opts.serverId, inName, {
+        natSrc: opts.natSrc,
+        natDst: opts.natDst,
+        natSrcPort: opts.natSrcPort,
+        natDstPort: opts.natDstPort,
+      }),
+    )
+  if (!dest) return false
+  const ours = flowOursHosts(opts.topo)
+  if (isLocalIp(dest, ours) || ours.has(dest)) return false
  if (opts.serverType === "exit-node" && opts.topo) {
    const client =
      resolveClient(opts.topo, opts.serverId, inName)
@@ -3,6 +3,7 @@ import { dbQuery } from "../db/index.js"
 import { withPgOrSkip } from "../test/pg.js"
 import { ensurePartitionFor } from "../db/partitions.js"
 import { pool } from "../db/index.js"
+import { applySqlMigrations } from "../db/migrate.js"
 import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js"
 import {
  disableRipeEnqueueForTests,
@@ -19,6 +20,8 @@ if (!(await withPgOrSkip())) {
  process.exit(0)
 }

+await applySqlMigrations(pool)
+
 const nServers = (await dbQuery<{ n: number }>(`SELECT COUNT(*)::int AS n FROM servers`)).rows[0]?.n ?? 0
 if (nServers > 10) {
  console.warn("traffic-flow-facts-rebuild.test.ts: skip (не пустая БД)")
@@ -119,10 +122,10 @@ try {
  `, [serverId])
  const byAsn = new Map(rows.rows.map((r) => [Number(r.asn), Number(r.bytes)]))
  const total = [...byAsn.values()].reduce((s, n) => s + n, 0)
-  assert.equal(total, 150)
+  assert.equal(total, 50)
  assert.equal(byAsn.get(12389), undefined, "ASN клиента не в hour facts")
  assert.equal(byAsn.get(15169), 50)
-  assert.equal(byAsn.get(0), 100)
+  assert.equal(byAsn.get(0), undefined)
 } finally {
  await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id = $1`, [serverId])
  await dbQuery(`DELETE FROM flow_daily_facts WHERE server_id = $1`, [serverId])
@@ -69,10 +69,16 @@ export async function rebuildFlowFactsFromBuckets(): Promise<FlowFactsRebuildRes
        packets: number
        inIface: string
        outIface: string
+        natSrc: string
+        natDst: string
+        natSrcPort: number
+        natDstPort: number
      }>(`
        SELECT server_id AS "serverId", bucket_at AS "bucketAt",
          host(src) AS src, host(dst) AS dst, proto, src_port AS "srcPort", dst_port AS "dstPort",
-          bytes, packets, in_iface AS "inIface", COALESCE(out_iface, '') AS "outIface"
+          bytes, packets, in_iface AS "inIface", COALESCE(out_iface, '') AS "outIface",
+          COALESCE(host(nat_src), '') AS "natSrc", COALESCE(host(nat_dst), '') AS "natDst",
+          COALESCE(nat_src_port, 0) AS "natSrcPort", COALESCE(nat_dst_port, 0) AS "natDstPort"
        FROM flow_buckets
        ORDER BY bucket_at, server_id
        LIMIT ? OFFSET ?
@@ -81,6 +87,20 @@ export async function rebuildFlowFactsFromBuckets(): Promise<FlowFactsRebuildRes
      for (const row of rows) {
        buckets += 1
        const serverType = catalog.byId.get(row.serverId)?.type
+        const destMeta = resolveInternetDest({
+          src: row.src,
+          dst: row.dst,
+          proto: Number(row.proto) || 0,
+          srcPort: Number(row.srcPort) || 0,
+          dstPort: Number(row.dstPort) || 0,
+          serverId: row.serverId,
+          inIface: row.inIface,
+          topo,
+          natSrc: row.natSrc,
+          natDst: row.natDst,
+          natSrcPort: Number(row.natSrcPort) || 0,
+          natDstPort: Number(row.natDstPort) || 0,
+        })
        if (!shouldWriteFlowFact({
          serverId: row.serverId,
          serverType,
@@ -92,17 +112,12 @@ export async function rebuildFlowFactsFromBuckets(): Promise<FlowFactsRebuildRes
          src: row.src,
          dst: row.dst,
          topo,
+          dest: destMeta.dest,
+          natSrc: row.natSrc,
+          natDst: row.natDst,
+          natSrcPort: Number(row.natSrcPort) || 0,
+          natDstPort: Number(row.natDstPort) || 0,
        })) continue
-        const destMeta = resolveInternetDest({
-          src: row.src,
-          dst: row.dst,
-          proto: Number(row.proto) || 0,
-          srcPort: Number(row.srcPort) || 0,
-          dstPort: Number(row.dstPort) || 0,
-          serverId: row.serverId,
-          inIface: row.inIface,
-          topo,
-        })
        bumpFlowFact({
          serverId: row.serverId,
          bucketAt: hourFromBucket(row.bucketAt),
@@ -21,6 +21,7 @@ import {
  minuteDimsSnapshotForTests,
 } from "./traffic-flow-engine.js"
 import { classifyFlowDst } from "./traffic-flow-classify.js"
+import { mapInternetBrand } from "./traffic-flow-dest.js"
 import { disableGeoipDbForTests } from "./geoip-settings.js"
 import {
  collectGeoipUpdateOnce,
@@ -102,6 +103,21 @@ assert.equal(lookupGeoip("6.6.6.6")?.country, "US")
 const classified = classifyFlowDst("8.8.8.8", 6, 443, 51504, hit)
 assert.equal(classified.service, "Google")

+setGeoipReadersForTests({
+  country: fakeCountryReader({ "8.8.8.8": "US", "2001:4860:4860::8888": "US" }),
+  asn: fakeAsnReader({
+    "8.8.8.8": { asn: 15169, org: "GOOGLE" },
+    "2001:4860:4860::8888": { asn: 15169, org: "GOOGLE" },
+    "64.233.161.1": { asn: 15169, org: "GOOGLE" },
+  }),
+})
+const v6meta = resolveFlowIp("2001:4860:4860::8888")
+assert.equal(v6meta?.asn, 15169)
+assert.equal(mapInternetBrand("2001:4860:4860::8888", 17, 443, 50000, v6meta).service, "YouTube")
+assert.notEqual(mapInternetBrand("2001:4860:4860::8888", 17, 443, 50000, v6meta).service, "Прочее")
+const cidrYt = mapInternetBrand("64.233.161.1", 17, 443, 50000, resolveFlowIp("64.233.161.1"))
+assert.equal(cidrYt.service, "YouTube")
+
 // ── движок: dims country/asn наполняются из geoip-ридеров ────────────────────
 resetEngineForTests()
 ingestParsedFlowsForServerForTests(1, [{
@@ -15,6 +15,7 @@ import {
  lastFlushUsedTransactionForTests,
  maybeRefreshIfaces,
  peekPendingFlows,
+  capFlowRowsPerServerBucket,
  resetFlowRingsForTests,
  setPendingCapForTests,
  setRefreshIfacesForTests,
@@ -146,4 +147,20 @@ resetFlowRingsForTests()
 resetIfaceCacheForTests()
 setRefreshIfacesForTests(null)

+{
+  const minute0 = "2026-01-01T00:00:00.000Z"
+  const minute1 = "2026-01-01T00:01:00.000Z"
+  const rows: Array<{ serverId: number; bucketAt: string; bytes: number; id: string }> = []
+  for (let i = 1; i <= 25; i++) {
+    rows.push({ serverId: 1, bucketAt: minute0, bytes: i, id: `a${i}` })
+    rows.push({ serverId: 2, bucketAt: minute0, bytes: i, id: `b${i}` })
+  }
+  rows.push({ serverId: 1, bucketAt: minute1, bytes: 1, id: "a-min-other-minute" })
+  const capped = capFlowRowsPerServerBucket(rows, 20)
+  assert.equal(capped.filter((r) => r.serverId === 1 && r.bucketAt === minute0).length, 20)
+  assert.equal(capped.filter((r) => r.serverId === 2).length, 20)
+  assert.ok(capped.some((r) => r.id === "a-min-other-minute"), "другая минута не режется глобальным top-N")
+  assert.ok(!capped.some((r) => r.id === "a1" || r.id === "b1"), "мелкие 5-tuple сервера выпадают только в своём bucket")
+}
+
 console.log("traffic-flow-ingest.test.ts: ok")
@@ -1,8 +1,7 @@
 import { Worker } from "node:worker_threads"
-import { gte, sql } from "drizzle-orm"
-import { db, dbGet, dbQuery, pool, withAdvisoryLock } from "../db/index.js"
+import { db, dbAll, dbGet, dbQuery, pool, withAdvisoryLock } from "../db/index.js"
 import { dropExpiredPartitions } from "../db/partitions.js"
-import { flowBuckets, servers } from "../db/schema.js"
+import { servers } from "../db/schema.js"
 import type { FlowPurgeDto, FlowStatsDto, FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
 import { protoName, type ParsedFlowInput } from "./traffic-flow-parse.js"
 import type { CollectorHeartbeat, ExporterMapPayload, MainToWorker, WorkerToMain } from "./traffic-flow-collector-ipc.js"
@@ -288,6 +287,10 @@ function mergeInto(map: Map<string, PendingFlowRow>, row: PendingFlowRow): void
    prev.packets += row.packets
    if (row.outIface && !prev.outIface) prev.outIface = row.outIface
    if (row.nextHop && !prev.nextHop) prev.nextHop = row.nextHop
+    if (row.natSrc && !prev.natSrc) prev.natSrc = row.natSrc
+    if (row.natDst && !prev.natDst) prev.natDst = row.natDst
+    if (row.natSrcPort && !prev.natSrcPort) prev.natSrcPort = row.natSrcPort
+    if (row.natDstPort && !prev.natDstPort) prev.natDstPort = row.natDstPort
    if (row.flowStartMs && (!prev.flowStartMs || row.flowStartMs < prev.flowStartMs)) prev.flowStartMs = row.flowStartMs
    if (row.flowEndMs > (prev.flowEndMs ?? 0)) prev.flowEndMs = row.flowEndMs
    return
@@ -295,6 +298,32 @@ function mergeInto(map: Map<string, PendingFlowRow>, row: PendingFlowRow): void
  map.set(key, { ...row })
 }

+/** Top-N разговоров на (server_id, bucket_at), как prune persist — не глобальный ORDER BY bytes. */
+export function capFlowRowsPerServerBucket<T extends { serverId: number; bucketAt: string; bytes: number }>(
+  rows: T[],
+  keep: number,
+): T[] {
+  const cap = Math.max(20, keep)
+  const groups = new Map<string, T[]>()
+  for (const row of rows) {
+    const k = `${row.serverId}\0${row.bucketAt}`
+    const list = groups.get(k)
+    if (list) list.push(row)
+    else groups.set(k, [row])
+  }
+  const out: T[] = []
+  for (const list of groups.values()) {
+    list.sort((a, b) => b.bytes - a.bytes)
+    out.push(...list.slice(0, cap))
+  }
+  return out
+}
+
+function isoBucketAt(v: unknown): string {
+  if (v instanceof Date) return v.toISOString()
+  return String(v ?? "")
+}
+
 export async function listLiveFlowRows(sinceIso: string): Promise<PendingFlowRow[]> {
  if (worker && lastHeartbeat?.workerAlive) {
    return await listStoredFlowRows(sinceIso)
@@ -302,30 +331,67 @@ export async function listLiveFlowRows(sinceIso: string): Promise<PendingFlowRow
  return engineListLive(sinceIso)
 }

+interface StoredBucketRow {
+  server_id: number
+  bucket_at: string | Date
+  src: string
+  dst: string
+  proto: number
+  src_port: number
+  dst_port: number
+  bytes: number
+  packets: number
+  in_iface: string
+  out_iface: string | null
+  next_hop: string | null
+  flow_start_ms: number | null
+  flow_end_ms: number | null
+  nat_src: string | null
+  nat_dst: string | null
+  nat_src_port: number | null
+  nat_dst_port: number | null
+}
+
 export async function listStoredFlowRows(sinceIso: string): Promise<PendingFlowRow[]> {
  const settings = await getTrafficFlowSettingsRow()
-  const cap = Math.max(20, settings.topN) * 60
-  const stored = await db.select().from(flowBuckets)
-    .where(gte(flowBuckets.bucketAt, sinceIso))
-    .orderBy(sql`${flowBuckets.bytes} DESC`)
-    .limit(cap)
+  const keep = Math.max(20, settings.topN)
+  const stored = await dbAll<StoredBucketRow>(`
+    SELECT
+      server_id, bucket_at, src::text AS src, dst::text AS dst, proto,
+      src_port, dst_port, bytes, packets, in_iface, out_iface,
+      next_hop::text AS next_hop, flow_start_ms, flow_end_ms,
+      nat_src::text AS nat_src, nat_dst::text AS nat_dst, nat_src_port, nat_dst_port
+    FROM (
+      SELECT fb.*,
+             ROW_NUMBER() OVER (
+               PARTITION BY server_id, bucket_at ORDER BY bytes DESC
+             ) AS rn
+      FROM flow_buckets fb
+      WHERE bucket_at >= $1
+    ) ranked
+    WHERE rn <= $2
+  `, [sinceIso, keep])
  const merged = new Map<string, PendingFlowRow>()
  for (const r of stored) {
    mergeInto(merged, {
-      serverId: r.serverId,
-      bucketAt: r.bucketAt,
+      serverId: Number(r.server_id),
+      bucketAt: isoBucketAt(r.bucket_at),
      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: r.outIface ?? "",
-      nextHop: r.nextHop ?? "",
-      flowStartMs: r.flowStartMs ?? 0,
-      flowEndMs: r.flowEndMs ?? 0,
+      proto: Number(r.proto) || 0,
+      srcPort: Number(r.src_port) || 0,
+      dstPort: Number(r.dst_port) || 0,
+      bytes: Number(r.bytes) || 0,
+      packets: Number(r.packets) || 0,
+      inIface: r.in_iface ?? "",
+      outIface: r.out_iface ?? "",
+      nextHop: r.next_hop ?? "",
+      flowStartMs: Number(r.flow_start_ms) || 0,
+      flowEndMs: Number(r.flow_end_ms) || 0,
+      natSrc: r.nat_src ?? "",
+      natDst: r.nat_dst ?? "",
+      natSrcPort: Number(r.nat_src_port) || 0,
+      natDstPort: Number(r.nat_dst_port) || 0,
    })
  }
  if (!worker) {
@@ -334,7 +400,7 @@ export async function listStoredFlowRows(sinceIso: string): Promise<PendingFlowR
      mergeInto(merged, p)
    }
  }
-  return [...merged.values()]
+  return capFlowRowsPerServerBucket([...merged.values()], keep)
 }

 export async function listFlowRowsForWindow(minutes: number): Promise<PendingFlowRow[]> {
@@ -1,5 +1,5 @@
 import assert from "node:assert/strict"
-import { isNonPublicIp, pickInternetDest, pickInternetPeer } from "./traffic-flow-ip.js"
+import { isNonPublicIp, pickInternetDest, pickInternetPeer, pickMapInternetDest } from "./traffic-flow-ip.js"

 assert.equal(isNonPublicIp("10.200.100.53"), true)
 assert.equal(isNonPublicIp("173.194.151.65"), false)
@@ -56,5 +56,51 @@ assert.equal(
  "8.8.8.8",
  "ours как dst: dest = публичный src",
 )
+assert.equal(
+  pickInternetDest("203.0.113.10", "198.51.100.1", 0, 0, { ours }),
+  "",
+  "JH ours → EN ours: dest нет",
+)
+assert.equal(
+  pickInternetDest("10.200.100.53", "10.200.100.1", 53880, 443, {
+    ...client,
+    natDst: "8.8.8.8",
+    natDstPort: 443,
+  }),
+  "8.8.8.8",
+  "RFC1918 + NAT Google",
+)
+assert.equal(
+  pickInternetDest("10.200.100.53", "10.200.100.1", 53880, 443, {
+    ...client,
+    natDst: "0.0.0.0",
+  }),
+  "",
+  "NAT 0.0.0.0 не dest",
+)
+assert.equal(
+  pickMapInternetDest("10.200.100.53", "10.200.100.1", 53880, 443, {
+    ...client,
+    natDst: "8.8.8.8",
+    natDstPort: 443,
+  }),
+  "8.8.8.8",
+  "карта: RFC1918 + NAT Google",
+)
+assert.equal(
+  pickMapInternetDest("10.200.100.53", "10.200.100.1", 53880, 443, client),
+  "",
+  "карта: RFC1918 без NAT → Прочее",
+)
+assert.equal(
+  pickMapInternetDest("173.194.151.65", "10.200.100.53", 12345, 57182, client),
+  "173.194.151.65",
+  "карта: реверс googlevideo не :443 — всё равно публичный src",
+)
+assert.equal(
+  pickMapInternetDest("203.0.113.10", "198.51.100.1", 0, 0, { ours }),
+  "",
+  "карта: JH ours → EN ours всё ещё не dest",
+)

 console.log("traffic-flow-ip.test.ts: ok")
@@ -55,57 +55,104 @@ export function isNonPublicIp(ip: string): boolean {

 const PEER_WELL_KNOWN_PORTS = new Set([80, 443, 53, 853])

+export function isUnspecifiedIp(ip: string): boolean {
+  const t = String(ip ?? "").trim()
+  if (!t) return true
+  const lower = t.toLowerCase()
+  return t === "0.0.0.0" || lower === "::" || lower === "::0"
+}
+
+function usableIp(ip: string | undefined): string {
+  const t = String(ip ?? "").trim()
+  return isUnspecifiedIp(t) ? "" : t
+}
+
 export interface InternetDestCtx {
  /** WAN IP узлов сети (EN/JH) — не интернет-назначение. */
  ours?: ReadonlySet<string>
  /** Ingress с bound GRE/WG клиента: dest = нелокальный IP, не ASN клиента. */
  boundClient?: boolean
+  /** IPFIX postNAT (IANA 225/226). */
+  natSrc?: string
+  natDst?: string
+  /** IPFIX postNAPT ports (IANA 227/228). */
+  natSrcPort?: number
+  natDstPort?: number
 }

 export function isLocalIp(ip: string, ours?: ReadonlySet<string>): boolean {
-  if (isNonPublicIp(ip)) return true
+  if (isUnspecifiedIp(ip) || isNonPublicIp(ip)) return true
  return Boolean(ours?.has(String(ip ?? "").trim()))
 }

 /**
 * Интернет-назначение потока для ASN/страны/сервиса.
- * Пустая строка — dest нет (не GeoIP IP клиента).
+ * Пустая строка — dest нет (не GeoIP IP клиента / GRE-пира).
 */
 export function pickInternetDest(
-  src: string,
-  dst: string,
+  srcRaw: string,
+  dstRaw: string,
  srcPort: number,
  dstPort: number,
  ctx?: InternetDestCtx,
 ): string {
  const ours = ctx?.ours
-  const srcLocal = isLocalIp(src, ours)
-  const dstLocal = isLocalIp(dst, ours)
-  const srcPub = !srcLocal
-  const dstPub = !dstLocal
+  const src = usableIp(srcRaw)
+  const dst = usableIp(dstRaw)
+  const natSrc = usableIp(ctx?.natSrc)
+  const natDst = usableIp(ctx?.natDst)
+  const internet = (ip: string) => Boolean(ip) && !isLocalIp(ip, ours)
+  const dstIp = internet(dst) ? dst : (internet(natDst) ? natDst : "")
+  const srcIp = internet(src) ? src : (internet(natSrc) ? natSrc : "")
+  const dstPortEff = internet(dst) ? dstPort : (internet(natDst) ? (ctx?.natDstPort || dstPort) : dstPort)
+  const srcPortEff = internet(src) ? srcPort : (internet(natSrc) ? (ctx?.natSrcPort || srcPort) : srcPort)

  if (ctx?.boundClient) {
-    if (dstPub) return dst
-    if (srcPub && dstLocal) {
-      const srcWk = PEER_WELL_KNOWN_PORTS.has(srcPort)
-      const dstWk = PEER_WELL_KNOWN_PORTS.has(dstPort)
-      if (srcWk && !dstWk) return src
+    if (dstIp) return dstIp
+    if (srcIp) {
+      const srcWk = PEER_WELL_KNOWN_PORTS.has(srcPortEff)
+      const dstWk = PEER_WELL_KNOWN_PORTS.has(dstPortEff)
+      if (srcWk && !dstWk) return srcIp
      return ""
    }
    return ""
  }

-  if (srcPub && !dstPub) return src
-  if (dstPub && !srcPub) return dst
-  if (srcPub && dstPub) {
-    const srcWk = PEER_WELL_KNOWN_PORTS.has(srcPort)
-    const dstWk = PEER_WELL_KNOWN_PORTS.has(dstPort)
-    if (srcWk && !dstWk) return src
-    if (dstWk && !srcWk) return dst
+  if (srcIp && !dstIp) return srcIp
+  if (dstIp && !srcIp) return dstIp
+  if (srcIp && dstIp) {
+    const srcWk = PEER_WELL_KNOWN_PORTS.has(srcPortEff)
+    const dstWk = PEER_WELL_KNOWN_PORTS.has(dstPortEff)
+    if (srcWk && !dstWk) return srcIp
+    if (dstWk && !srcWk) return dstIp
+    return dstIp
  }
+
+  if (src && dst && ours?.has(src) && ours.has(dst)) return ""
  return dst
 }

+/**
+ * Dest для карты: сначала pickInternetDest, иначе любой публичный src/dst/NAT
+ * (без well-known ловушки boundClient — реверс googlevideo не только :80/:443/:53).
+ */
+export function pickMapInternetDest(
+  srcRaw: string,
+  dstRaw: string,
+  srcPort: number,
+  dstPort: number,
+  ctx?: InternetDestCtx,
+): string {
+  const dest = pickInternetDest(srcRaw, dstRaw, srcPort, dstPort, ctx)
+  if (dest) return dest
+  const ours = ctx?.ours
+  const internet = (ip: string) => Boolean(ip) && !isLocalIp(ip, ours)
+  for (const ip of [usableIp(ctx?.natDst), usableIp(ctx?.natSrc), usableIp(dstRaw), usableIp(srcRaw)]) {
+    if (internet(ip)) return ip
+  }
+  return ""
+}
+
 /**
 * Интернет-сторона потока без топологии: у IPFIX сервис часто в src (Google:443 → RFC1918).
 * Для куба статистики используйте pickInternetDest.
@@ -70,6 +70,21 @@ import {
  console.log("traffic-flow-map-hops.test.ts: pickMapServices ok")
 }

+{
+  const leak = pickMapServices(
+    [
+      { id: "svc:other", label: "Прочее", category: "Прочее", bytes: 19_000, bps: 0, share: 19 / 30 },
+      { id: "svc:google", label: "Google", category: "Веб", bytes: 11_000, bps: 0, share: 11 / 30 },
+    ],
+    5,
+  )
+  assert.equal(leak.reduce((n, s) => n + s.bytes, 0), 30_000)
+  const google = leak.find((s) => s.id === "svc:google")
+  assert.ok(google)
+  assert.ok(google.share < 0.4, "доля от окна хопа, не от named-only")
+  assert.ok(leak.some((s) => s.id === "svc:other"), "дыра видна как Прочее")
+}
+
 if (!(await withPgOrSkip())) {
  console.log("traffic-flow-map-hops.test.ts: skip")
  process.exit(0)
@@ -279,12 +294,18 @@ try {
  resetFlowMapHopsCacheForTests()
  const six = await buildFlowMapHops({ minutes: 5, minSharePct: 5 })
  assert.equal(six.totalBytes, 10_000)
+  assert.equal(six.namedBytes, 600)
+  assert.equal(six.unclassifiedBytes, 9400)
  const google = six.services?.find((s) => s.id === "svc:google")
+  const otherSix = six.services?.find((s) => s.id === "svc:other")
  assert.ok(google, "Google ≥ 5%")
-  assert.ok(google.share >= 0.05)
+  assert.ok(otherSix, "остаток — Прочее")
+  assert.ok(google.share >= 0.05 && google.share < 0.1)
+  assert.ok(otherSix.share >= 0.9)
  const googleEdge = six.serviceEdges?.find((e) => e.toId === "svc:google" && e.fromId === "9")
  assert.ok(googleEdge)
  assert.equal(googleEdge.clientName, "Alice")
+  assert.equal((six.serviceEdges ?? []).reduce((n, e) => n + e.bytes, 0), 10_000)
 } finally {
  resetFlowRingsForTests()
  resetIfaceCacheForTests()
@@ -309,9 +330,14 @@ try {
  resetFlowMapHopsCacheForTests()
  const four = await buildFlowMapHops({ minutes: 5, minSharePct: 5 })
  assert.equal(four.totalBytes, 10_000)
+  assert.equal(four.namedBytes, 400)
+  assert.equal(four.unclassifiedBytes, 9600)
  const googleFour = four.services?.find((s) => s.id === "svc:google")
-  assert.ok(googleFour, "единственный бренд виден при 4% от окна")
-  assert.ok(googleFour.share >= 0.99, "доля среди брендов ≈ 1")
+  const otherFour = four.services?.find((s) => s.id === "svc:other")
+  assert.ok(googleFour, "бренд виден при 4% от окна (MIN_NODES)")
+  assert.ok(otherFour, "Прочее держит остаток окна")
+  assert.ok(googleFour.share < 0.1, "доля от totalBytes, не от named")
+  assert.ok(otherFour.share > 0.9)
  resetFlowMapHopsCacheForTests()
  const off = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
  assert.ok(off.services?.some((s) => s.id === "svc:google"), "порог 0 показывает Google")
@@ -340,10 +366,13 @@ try {
  const two = await buildFlowMapHops({ minutes: 5, minSharePct: 5 })
  const googleTwo = two.services?.find((s) => s.id === "svc:google")
  const cfTwo = two.services?.find((s) => s.id === "svc:cloudflare")
+  const otherTwo = two.services?.find((s) => s.id === "svc:other")
  assert.ok(googleTwo, "Google среди брендов")
  assert.ok(cfTwo, "Cloudflare среди брендов")
-  assert.ok(googleTwo.share >= 0.05)
-  assert.ok(cfTwo.share >= 0.05)
+  assert.ok(otherTwo, "Прочее")
+  assert.ok(googleTwo.share > 0.03 && googleTwo.share < 0.05)
+  assert.ok(cfTwo.share > 0.03 && cfTwo.share < 0.05)
+  assert.ok(otherTwo.share > 0.9)
 } finally {
  resetFlowRingsForTests()
  resetIfaceCacheForTests()
@@ -373,6 +402,43 @@ try {
  resetRipeCacheForTests()
 }

+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+seedFlowTopologyForTests(topo)
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "gre-client" },
+  { ".id": "*3", name: "gre-jh-en" },
+])
+ingestParsedFlowsForServerForTests(7, [
+  payloadFlow("64.233.161.1", 10_000),
+  payloadFlow("203.0.113.50", 20_000),
+])
+try {
+  resetFlowMapHopsCacheForTests()
+  const leak = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
+  const hop = leak.hops.find((h) => h.kind === "gre" && h.fromId === "7" && h.toId === "9")
+  assert.ok(hop, "GRE hop JH→EN")
+  assert.equal(hop.bytes, 30_000)
+  assert.equal(leak.totalBytes, 30_000)
+  assert.equal(leak.namedBytes, 10_000)
+  assert.equal(leak.unclassifiedBytes, 20_000)
+  const yt = leak.services?.find((s) => s.id === "svc:youtube")
+  const other = leak.services?.find((s) => s.id === "svc:other")
+  assert.ok(yt, "64.233:443 без RIPE → YouTube")
+  assert.ok(other, "dest без бренда → Прочее")
+  assert.equal(yt.bytes, 10_000)
+  assert.equal(other.bytes, 20_000)
+  assert.ok(Math.abs(yt.share - 10 / 30) < 0.01)
+  assert.ok(Math.abs(other.share - 20 / 30) < 0.01)
+  assert.equal((leak.serviceEdges ?? []).reduce((n, e) => n + e.bytes, 0), hop.bytes)
+} finally {
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+}
+
 const smallBrands: Array<{ ip: string; asn: number; holder: string; bytes: number; id: string }> = [
  { ip: "203.0.113.1", asn: 714, holder: "APPLE-ENGINEERING", bytes: 400, id: "svc:apple" },
  { ip: "203.0.113.2", asn: 36459, holder: "GITHUB", bytes: 390, id: "svc:github" },
@@ -450,6 +516,72 @@ try {
  resetFlowCatalogForTests()
 }

+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+seedFlowTopologyForTests(topo)
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "gre-client" },
+  { ".id": "*3", name: "gre-jh-en" },
+])
+googleRipe()
+ingestParsedFlowsForServerForTests(7, [
+  {
+    src: "10.100.1.17",
+    dst: "8.8.8.8",
+    proto: 6,
+    srcPort: 51234,
+    dstPort: 443,
+    bytes: 4_000,
+    packets: 10,
+    inIface: "2",
+    outIface: "3",
+    nextHop: "198.51.100.1",
+  },
+  {
+    src: "203.0.113.10",
+    dst: "198.51.100.1",
+    proto: 47,
+    srcPort: 0,
+    dstPort: 0,
+    bytes: 2_000_000,
+    packets: 400,
+    inIface: "3",
+    outIface: "3",
+  },
+  {
+    src: "10.200.100.53",
+    dst: "10.200.100.1",
+    proto: 6,
+    srcPort: 53880,
+    dstPort: 443,
+    bytes: 3_000,
+    packets: 8,
+    inIface: "2",
+    outIface: "3",
+    nextHop: "198.51.100.1",
+    natDst: "8.8.8.8",
+    natDstPort: 443,
+  },
+])
+try {
+  resetFlowMapHopsCacheForTests()
+  const path = await buildFlowMapHops({ minutes: 5, excludeOverlay: false, excludeMesh: false, minSharePct: 0 })
+  const hop = path.hops.find((h) => h.kind === "gre" && h.fromId === "7" && h.toId === "9")
+  assert.ok(hop, "hop JH→EN")
+  const google = path.services?.find((s) => s.id === "svc:google")
+  assert.ok(google, "сервис Google")
+  assert.equal(google.bytes, 7_000)
+  assert.ok(!(path.services ?? []).some((s) => s.label === "GRE"), "GRE не dest")
+} finally {
+  seedFlowTopologyForTests(null)
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
 resetFlowRingsForTests()
 resetIfaceCacheForTests()
 resetRipeCacheForTests()
@@ -488,9 +620,9 @@ ingestParsedFlowsForServerForTests(7, [
 try {
  resetFlowMapHopsCacheForTests()
  const rev = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
-  assert.ok(rev.services?.some((s) => s.id === "svc:google"), "реверс Google:443 → 10.x")
+  assert.ok(rev.services?.some((s) => s.id === "svc:youtube"), "реверс googlevideo:443 → YouTube")
  assert.ok(rev.services?.some((s) => s.id === "svc:cloudflare"), "реверс Cloudflare:443 → 10.x")
-  assert.ok(rev.serviceEdges?.some((e) => e.toId === "svc:google" && e.fromId === "9"))
+  assert.ok(rev.serviceEdges?.some((e) => e.toId === "svc:youtube" && e.fromId === "9"))
 } finally {
  seedFlowTopologyForTests(null)
  resetFlowRingsForTests()
@@ -566,13 +698,13 @@ ingestParsedFlowsForServerForTests(7, [
 try {
  resetFlowMapHopsCacheForTests()
  const wanOnly = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
-  const googleEdge = wanOnly.serviceEdges?.find((e) => e.toId === "svc:google")
-  assert.ok(googleEdge, "Google WAN без GRE payload")
+  const googleEdge = wanOnly.serviceEdges?.find((e) => e.toId === "svc:youtube")
+  assert.ok(googleEdge, "YouTube WAN без GRE payload")
  assert.equal(googleEdge.fromId, "9", "единственный EN, даже без nextHop")
  assert.ok(googleEdge.bps > 0, "скорость на hop EN→сервис")
  assert.ok(!(wanOnly.serviceEdges ?? []).some((e) => e.fromId === "7"), "нет пунктира с JH")
-  const googlePath = wanOnly.servicePaths?.find((p) => p.serviceId === "svc:google")
-  assert.ok(googlePath, "путь WAN Google")
+  const googlePath = wanOnly.servicePaths?.find((p) => p.serviceId === "svc:youtube")
+  assert.ok(googlePath, "путь WAN YouTube")
  assert.equal(googlePath.viaId, "7", "via = JH exporter")
  assert.equal(googlePath.enId, "9", "якорь EN")
  assert.ok(googlePath.bps > 0, "скорость на пути клиента")
@@ -2,20 +2,16 @@ import { eq } from "drizzle-orm"
 import type { FlowMapHop, FlowMapHopsDto, FlowMapService, FlowMapServiceEdge, FlowMapServicePath } from "@mmapp/contracts/traffic-flow"
 import { db } from "../db/index.js"
 import { userInterfaceBindings } from "../db/schema.js"
-import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
-import {
-  isNamedInternetService,
-  mapServiceNodeId,
-  resolveFlowBrand,
-} from "./traffic-flow-brands.js"
+import { flowRowMatchesFilter } from "./traffic-flow-apps.js"
+import { OTHER_SERVICE, isNamedInternetService, mapServiceNodeId } from "./traffic-flow-brands.js"
+import { refreshFlowCatalogInBackground } from "./traffic-flow-classify.js"
 import { dedupFlowRowsAcrossExporters, dedupFlowRowsMaxBytes } from "./traffic-flow-dedup.js"
 import { getFlowListenerState, listFlowRowsForWindow } from "./traffic-flow-ingest.js"
 import { resolveIfaceName } from "./traffic-flow-ifaces.js"
 import { classifyFlowPlane, shouldKeepPlane } from "./traffic-flow-planes.js"
-import { destCtxForIface } from "./traffic-flow-dest.js"
-import { pickInternetDest } from "./traffic-flow-ip.js"
-import { type FlowIpMeta } from "./traffic-flow-ripe.js"
-import { resolveFlowIp } from "./traffic-flow-geoip.js"
+import { destCtxForIface, mapInternetBrand } from "./traffic-flow-dest.js"
+import { pickMapInternetDest } from "./traffic-flow-ip.js"
+import { geoipReadersStatus, resolveFlowIp } from "./traffic-flow-geoip.js"
 import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js"
 import { loadFlowTopology, resolveClient, resolveEn, getServerCatalog, type FlowTopology } from "./traffic-flow-topology.js"
 import { flowDataEpoch } from "./traffic-flow-engine.js"
@@ -103,7 +99,7 @@ export function clampMapServiceMinSharePct(n: unknown): number {
  return Math.min(100, Math.max(0, v))
 }

-/** Доля среди именованных брендов; порог ИЛИ топ-N, затем cap. */
+/** Доля от payload окна; порог ИЛИ топ-N, затем cap. */
 export function pickMapServices(ranked: FlowMapService[], minSharePct: number): FlowMapService[] {
  if (minSharePct <= 0) return ranked.slice(0, MAP_SERVICE_NODE_CAP)
  const minShare = minSharePct / 100
@@ -186,22 +182,6 @@ function toHop(a: HopAcc, windowSec: number): FlowMapHop {
  }
 }

-/** Имя бренда без каталога EvoBGP — только ASN/CIDR кэш + proto. */
-function classifyMapDstLite(
-  dst: string,
-  proto: number,
-  dstPort: number,
-  srcPort: number,
-  ripe: FlowIpMeta | null,
-): { service: string; category: string } | null {
-  if (proto === 47 || proto === 50) return null
-  const app = applicationName(proto, dstPort, srcPort)
-  if (app === "WireGuard" || app === "DNS" || app === "SSH" || app === "BGP") return null
-  const brand = resolveFlowBrand(dst, ripe?.asn ?? 0, ripe?.holder ?? "", proto, dstPort, srcPort)
-  if (!brand || !isNamedInternetService(brand.service, brand.category)) return null
-  return brand
-}
-
 async function resolveMinSharePct(q: FlowMapHopsQuery): Promise<number> {
  if (q.minSharePct != null) return clampMapServiceMinSharePct(q.minSharePct)
  try {
@@ -213,6 +193,7 @@ async function resolveMinSharePct(q: FlowMapHopsQuery): Promise<number> {
 }

 async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number): Promise<FlowMapHopsDto> {
+  refreshFlowCatalogInBackground()
  const windowSec = Math.max(60, q.minutes * 60)
  const raw = await listFlowRowsForWindow(q.minutes)
  const allow = q.userId ? await userIfaceAllow(q.userId) : null
@@ -355,16 +336,21 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
    const inName = resolveIfaceName(r.serverId, r.inIface).name
    const outName = resolveIfaceName(r.serverId, r.outIface).name
    totalBytes += r.bytes
-    const dest = pickInternetDest(
+    const dest = pickMapInternetDest(
      r.src,
      r.dst,
      r.srcPort,
      r.dstPort,
-      destCtxForIface(topo, r.serverId, inName),
+      destCtxForIface(topo, r.serverId, inName, {
+        natSrc: r.natSrc,
+        natDst: r.natDst,
+        natSrcPort: r.natSrcPort,
+        natDstPort: r.natDstPort,
+      }),
    )
-    if (!dest) continue
+    const destKey = dest || "__other__"
    const client = resolveMapClient(topo, r.serverId, inName, outName)
-    const prevDst = dstAcc.get(dest)
+    const prevDst = dstAcc.get(destKey)
    if (prevDst) {
      prevDst.bytes += r.bytes
      bumpFrom(prevDst, String(r.serverId), r.bytes, client)
@@ -377,7 +363,7 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
        fromBytes: new Map(),
      }
      bumpFrom(acc, String(r.serverId), r.bytes, client)
-      dstAcc.set(dest, acc)
+      dstAcc.set(destKey, acc)
    }
  }

@@ -432,9 +418,10 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
  }

  for (const [dst, acc] of dstAcc) {
-    const ripe = resolveFlowIp(dst)
-    const classified = classifyMapDstLite(dst, acc.proto, acc.dstPort, acc.srcPort, ripe)
-    if (!classified) continue
+    const ripe = dst && dst !== "__other__" ? resolveFlowIp(dst) : null
+    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
@@ -490,7 +477,11 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
    }
  }

-  const namedBytes = [...svcTotals.values()].reduce((n, s) => n + s.bytes, 0)
+  const namedBytes = [...svcTotals.values()]
+    .filter((s) => isNamedInternetService(s.label, s.category))
+    .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]) => ({
@@ -499,7 +490,7 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
        category: s.category,
        bytes: s.bytes,
        bps: (s.bytes * 8) / windowSec,
-        share: namedBytes > 0 ? s.bytes / namedBytes : 0,
+        share: shareBase > 0 ? s.bytes / shareBase : 0,
      }))
      .sort((a, b) => b.bytes - a.bytes),
    minSharePct,
@@ -539,6 +530,7 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
    .sort((a, b) => b.bps - a.bps)

  const listener = getFlowListenerState()
+  const geo = geoipReadersStatus()
  return {
    hops: [...hops.values()]
      .map((a) => toHop(a, windowSec))
@@ -547,6 +539,10 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
    rangeMinutes: q.minutes,
    windowSec,
    totalBytes,
+    namedBytes,
+    unclassifiedBytes,
+    asnLoaded: geo.asnLoaded,
+    countryLoaded: geo.countryLoaded,
    services,
    serviceEdges,
    servicePaths,
@@ -114,6 +114,8 @@ async function ensureIpfixFields(client: MikrotikClient): Promise<void> {
    "last-forwarded": "yes",
    "nat-src-address": "yes",
    "nat-dst-address": "yes",
+    "nat-src-port": "yes",
+    "nat-dst-port": "yes",
  })
  const rows = asRosArray<Record<string, unknown>>(await client.get("/ip/traffic-flow/ipfix"))
  const id = rows[0] ? rosRowId(rows[0]) : ""
@@ -124,6 +126,9 @@ async function ensureIpfixFields(client: MikrotikClient): Promise<void> {
  await client.post("/ip/traffic-flow/ipfix/set", body)
 }

+/** Максимум flows в RAM (docs: overflow обрезает новые 5-tuple). */
+export const FLOW_CACHE_ENTRIES = "256k"
+
 async function ensureTrafficFlow(
  client: MikrotikClient,
  collectorIp: string,
@@ -132,6 +137,7 @@ async function ensureTrafficFlow(
  const body = toRosBody({
    enabled: "yes",
    interfaces: "all",
+    "cache-entries": FLOW_CACHE_ENTRIES,
    "active-flow-timeout": "1m",
    "inactive-flow-timeout": "15s",
  })
@@ -165,6 +171,23 @@ async function ensureTrafficFlow(
  await client.put("/ip/traffic-flow/target", targetBody)
 }

+/** GRE allow-fast-path=no: inner пакеты идут через CPU и попадают в Traffic Flow. Нагрузка на CPU. */
+export async function ensureGreSlowPath(client: MikrotikClient): Promise<number> {
+  const rows = asRosArray<Record<string, unknown>>(await client.get("/interface/gre"))
+  let patched = 0
+  for (const row of rows) {
+    const id = rosRowId(row)
+    if (!id) continue
+    const current = String(row["allow-fast-path"] ?? "true").toLowerCase()
+    if (current === "false" || current === "no") continue
+    await patchRosPath(client, `/interface/gre/${encodeRosId(id)}`, toRosBody({
+      "allow-fast-path": "no",
+    }))
+    patched += 1
+  }
+  return patched
+}
+
 export function usablePublicHost(raw: string | undefined): string {
  if (!raw) return ""
  const host = raw.split(",")[0]?.trim().replace(/^\[/, "").replace(/\]:\d+$/, "").split(":")[0]?.trim() ?? ""
@@ -178,7 +201,7 @@ export function usablePublicHost(raw: string | undefined): string {

 export async function applyFlowOverlay(
  serverIdRaw: string | number,
-  opts?: { publicEndpoint?: string; requestHost?: string },
+  opts?: { publicEndpoint?: string; requestHost?: string; disableGreFastPath?: boolean },
 ): Promise<TrafficFlowOverlayResult> {
  const steps: string[] = []
  const keys = await ensureHostKeys()
@@ -273,7 +296,20 @@ export async function applyFlowOverlay(
    }

    await ensureTrafficFlow(client, settings.collectorIp, settings.flowListenPort)
-    steps.push(`Traffic Flow → ${settings.collectorIp}:${settings.flowListenPort} ipfix (src auto)`)
+    steps.push(`Traffic Flow → ${settings.collectorIp}:${settings.flowListenPort} ipfix (src auto, cache ${FLOW_CACHE_ENTRIES})`)
+
+    if (opts?.disableGreFastPath) {
+      try {
+        const n = await ensureGreSlowPath(client)
+        steps.push(
+          n > 0
+            ? `GRE allow-fast-path=no (${n}) — inner IPFIX через CPU`
+            : "GRE already allow-fast-path=no",
+        )
+      } catch {
+        steps.push("GRE allow-fast-path не изменён (нет /interface/gre)")
+      }
+    }

    const listed = await listWireGuardInterfaces({ serverId: String(server.id), includePrivateKey: false })
    const created = listed.interfaces.find((i) => i.name === IFACE_NAME)
@@ -1,6 +1,6 @@
 import assert from "node:assert/strict"
 import { parseFlowPacket, protoName, resetFlowTemplatesForTests, templateExporterCountForTests } from "./traffic-flow-parse.js"
-import { allocateOverlayAddress, FLOW_TARGET_SRC_AUTO, usablePublicHost } from "./traffic-flow-overlay.js"
+import { allocateOverlayAddress, FLOW_CACHE_ENTRIES, FLOW_TARGET_SRC_AUTO, usablePublicHost } from "./traffic-flow-overlay.js"

 function netflowV5One(): Buffer {
  const buf = Buffer.alloc(24 + 48)
@@ -38,6 +38,7 @@ 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")
+assert.equal(FLOW_CACHE_ENTRIES, "256k")

 resetFlowTemplatesForTests()
 {
@@ -182,6 +183,95 @@ resetFlowTemplatesForTests()
  assert.equal(extra[0]?.bytes, 1500)
 }

+resetFlowTemplatesForTests()
+{
+  const fieldSpecs: Array<[number, number]> = [
+    [8, 4],
+    [12, 4],
+    [225, 4],
+    [226, 4],
+    [227, 2],
+    [228, 2],
+    [1, 4],
+  ]
+  const tplSetLen = 4 + 4 + fieldSpecs.length * 4
+  const tpl = Buffer.alloc(16 + tplSetLen)
+  tpl.writeUInt16BE(10, 0)
+  tpl.writeUInt16BE(tpl.length, 2)
+  tpl.writeUInt16BE(2, 16)
+  tpl.writeUInt16BE(tplSetLen, 18)
+  tpl.writeUInt16BE(256, 20)
+  tpl.writeUInt16BE(fieldSpecs.length, 22)
+  let off = 24
+  for (const [type, len] of fieldSpecs) {
+    tpl.writeUInt16BE(type, off)
+    tpl.writeUInt16BE(len, off + 2)
+    off += 4
+  }
+  const recLen = fieldSpecs.reduce((n, [, len]) => n + len, 0)
+  const data = Buffer.alloc(16 + 4 + recLen)
+  data.writeUInt16BE(10, 0)
+  data.writeUInt16BE(data.length, 2)
+  data.writeUInt16BE(256, 16)
+  data.writeUInt16BE(4 + recLen, 18)
+  let d = 20
+  data[d] = 10; data[d + 1] = 200; data[d + 2] = 100; data[d + 3] = 53; d += 4
+  data[d] = 10; data[d + 1] = 200; data[d + 2] = 100; data[d + 3] = 1; d += 4
+  data[d] = 0; data[d + 1] = 0; data[d + 2] = 0; data[d + 3] = 0; d += 4
+  data[d] = 8; data[d + 1] = 8; data[d + 2] = 8; data[d + 3] = 8; d += 4
+  data.writeUInt16BE(53880, d); d += 2
+  data.writeUInt16BE(443, d); d += 2
+  data.writeUInt32BE(900, d)
+  parseFlowPacket(tpl, "10.255.254.9")
+  const nat = parseFlowPacket(data, "10.255.254.9")
+  assert.equal(nat.length, 1)
+  assert.equal(nat[0]?.src, "10.200.100.53")
+  assert.equal(nat[0]?.dst, "10.200.100.1")
+  assert.equal(nat[0]?.natSrc, "0.0.0.0")
+  assert.equal(nat[0]?.natDst, "8.8.8.8")
+  assert.equal(nat[0]?.natSrcPort, 53880)
+  assert.equal(nat[0]?.natDstPort, 443)
+  assert.equal(nat[0]?.bytes, 900)
+}
+
+resetFlowTemplatesForTests()
+{
+  const fieldSpecs: Array<[number, number]> = [
+    [225, 4],
+    [12, 4],
+    [1, 4],
+  ]
+  const tplSetLen = 4 + 4 + fieldSpecs.length * 4
+  const tpl = Buffer.alloc(16 + tplSetLen)
+  tpl.writeUInt16BE(10, 0)
+  tpl.writeUInt16BE(tpl.length, 2)
+  tpl.writeUInt16BE(2, 16)
+  tpl.writeUInt16BE(tplSetLen, 18)
+  tpl.writeUInt16BE(256, 20)
+  tpl.writeUInt16BE(fieldSpecs.length, 22)
+  let off = 24
+  for (const [type, len] of fieldSpecs) {
+    tpl.writeUInt16BE(type, off)
+    tpl.writeUInt16BE(len, off + 2)
+    off += 4
+  }
+  const recLen = fieldSpecs.reduce((n, [, len]) => n + len, 0)
+  const data = Buffer.alloc(16 + 4 + recLen)
+  data.writeUInt16BE(10, 0)
+  data.writeUInt16BE(data.length, 2)
+  data.writeUInt16BE(256, 16)
+  data.writeUInt16BE(4 + recLen, 18)
+  let d = 20
+  data[d] = 0; data[d + 1] = 0; data[d + 2] = 0; data[d + 3] = 0; d += 4
+  data[d] = 8; data[d + 1] = 8; data[d + 2] = 8; data[d + 3] = 8; d += 4
+  data.writeUInt32BE(10, d)
+  parseFlowPacket(tpl, "10.255.254.10")
+  const zeroNat = parseFlowPacket(data, "10.255.254.10")
+  assert.equal(zeroNat[0]?.src, "")
+  assert.equal(zeroNat[0]?.natSrc, "0.0.0.0")
+  assert.equal(zeroNat[0]?.dst, "8.8.8.8")
+}
+
 resetFlowTemplatesForTests()
 {
  const tpl = Buffer.alloc(16 + 16 + 20)
@@ -13,6 +13,8 @@ export interface ParsedFlow {
  flowEndMs: number
  natSrc: string
  natDst: string
+  natSrcPort: number
+  natDstPort: number
 }

 export type ParsedFlowInput = Partial<ParsedFlow> & Pick<ParsedFlow, "src" | "dst" | "proto" | "bytes">
@@ -33,6 +35,8 @@ export function emptyParsedFlow(): ParsedFlow {
    flowEndMs: 0,
    natSrc: "",
    natDst: "",
+    natSrcPort: 0,
+    natDstPort: 0,
  }
 }

@@ -45,6 +49,8 @@ export function normalizeParsedFlow(flow: ParsedFlowInput): ParsedFlow {
    flowEndMs: flow.flowEndMs ?? 0,
    natSrc: flow.natSrc ?? "",
    natDst: flow.natDst ?? "",
+    natSrcPort: flow.natSrcPort ?? 0,
+    natDstPort: flow.natDstPort ?? 0,
    inIface: flow.inIface ?? "",
    outIface: flow.outIface ?? "",
    srcPort: flow.srcPort ?? 0,
@@ -86,6 +92,12 @@ function ipv4(buf: Buffer, offset: number): string {
  return `${buf[offset]}.${buf[offset + 1]}.${buf[offset + 2]}.${buf[offset + 3]}`
 }

+function usableIpfixIp(ip: string): boolean {
+  const t = String(ip ?? "").trim()
+  if (!t) return false
+  return t !== "0.0.0.0" && t.toLowerCase() !== "::" && t.toLowerCase() !== "::0"
+}
+
 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))
@@ -214,6 +226,8 @@ function recordFromFields(
  let flowEndMs = 0
  let natSrc = ""
  let natDst = ""
+  let natSrcPort = 0
+  let natDstPort = 0
  for (const f of fields) {
    const field = consumeField(buf, off, f.length, limit)
    if (!field) return null
@@ -243,15 +257,21 @@ function recordFromFields(
      case 225:
        if (data.length === 4) {
          natSrc = ipv4(data, 0)
-          if (!src) src = natSrc
+          if (!usableIpfixIp(src) && usableIpfixIp(natSrc)) src = natSrc
        }
        break
      case 226:
        if (data.length === 4) {
          natDst = ipv4(data, 0)
-          if (!dst) dst = natDst
+          if (!usableIpfixIp(dst) && usableIpfixIp(natDst)) dst = natDst
        }
        break
+      case 227:
+        natSrcPort = readUint(data, 0, data.length)
+        break
+      case 228:
+        natDstPort = readUint(data, 0, data.length)
+        break
      case 4:
        proto = readUint(data, 0, data.length)
        break
@@ -308,7 +328,7 @@ function recordFromFields(
  if (ifaceName && !inIface) inIface = ifaceName
  return {
    flow: normalizeParsedFlow({
-      src, dst, proto, srcPort, dstPort, bytes, packets, inIface, outIface, nextHop, flowStartMs, flowEndMs, natSrc, natDst,
+      src, dst, proto, srcPort, dstPort, bytes, packets, inIface, outIface, nextHop, flowStartMs, flowEndMs, natSrc, natDst, natSrcPort, natDstPort,
    }),
    next: off,
  }
@@ -2,7 +2,7 @@

 import { useEffect, useMemo, useState } from "react"
 import { toast } from "sonner"
-import { FormField } from "@/components/form-kit"
+import { FormField, FormToggle } from "@/components/form-kit"
 import { Alert, AlertDescription, AlertTitle } from "@/components/reui/alert"
 import { Frame, FramePanel } from "@/components/reui/frame"
 import { CodeBlock, downloadText } from "@/components/reui-kit/code-export-sheet"
@@ -54,12 +54,14 @@ function FlowOverlaySheet({
  const [result, setResult] = useState<TrafficFlowOverlayResult | null>(null)
  const [copied, setCopied] = useState(false)
  const [tab, setTab] = useState("linux")
+  const [disableGreFastPath, setDisableGreFastPath] = useState(false)

  useEffect(() => {
    if (!open) return
    setResult(null)
    setCopied(false)
    setTab("linux")
+    setDisableGreFastPath(false)
    const first = jumpHosts[0]
    const nextId = first ? String(first.id) : ""
    setServerId(nextId)
@@ -84,7 +86,9 @@ function FlowOverlaySheet({
    if (!serverId || !endpoint.trim()) return
    setBusy(true)
    try {
-      const res = await applyTrafficFlowOverlay(backendUrl, serverId, endpoint.trim())
+      const res = await applyTrafficFlowOverlay(backendUrl, serverId, endpoint.trim(), {
+        disableGreFastPath,
+      })
      setResult(res)
      setTab(res.hostFiles[0]?.id ?? "linux")
      toast.success(`wg-flow на ${res.address}`)
@@ -151,6 +155,15 @@ function FlowOverlaySheet({
              autoComplete="off"
            />
          </FormField>
+          <FormField
+            label="GRE slow-path (IPFIX inner)"
+            hint="allow-fast-path=no на GRE этого JH. Inner YouTube попадёт в Traffic Flow, но вырастет CPU. По умолчанию выкл."
+          >
+            <div className="flex items-center gap-2">
+              <FormToggle checked={disableGreFastPath} onChange={setDisableGreFastPath} />
+              <span className="text-sm text-muted-foreground">Выключить FastPath на GRE</span>
+            </div>
+          </FormField>
          {result ? (
            <div className="flex min-h-0 flex-col gap-4">
              <Alert variant="success">
@@ -47,6 +47,8 @@ export const trafficFlowSettingsPatchSchema = z.object({
 export const trafficFlowOverlayRequestSchema = z.object({
  serverId: z.union([z.string(), z.number()]),
  publicEndpoint: z.string().optional(),
+  /** GRE allow-fast-path=no: inner IPFIX через CPU. По умолчанию выкл (нагрузка на CPU). */
+  disableGreFastPath: z.boolean().optional(),
 })

 export const trafficFlowHostFileSchema = z.object({
@@ -316,6 +318,10 @@ export const flowMapHopsDtoSchema = z.object({
  rangeMinutes: z.number().int().positive(),
  windowSec: z.number().positive(),
  totalBytes: z.number().nonnegative().optional(),
+  namedBytes: z.number().nonnegative().optional(),
+  unclassifiedBytes: z.number().nonnegative().optional(),
+  asnLoaded: z.boolean().optional(),
+  countryLoaded: z.boolean().optional(),
  services: z.array(flowMapServiceDtoSchema).optional(),
  serviceEdges: z.array(flowMapServiceEdgeDtoSchema).optional(),
  servicePaths: z.array(flowMapServicePathDtoSchema).optional(),
@@ -46,10 +46,15 @@ export async function applyTrafficFlowOverlay(
  baseUrl: string,
  serverId: string | number,
  publicEndpoint?: string,
+  opts?: { disableGreFastPath?: boolean },
 ): Promise<TrafficFlowOverlayResult> {
  return requestJson(baseUrl, "/api/traffic/flow-overlay", {
    method: "POST",
-    body: JSON.stringify({ serverId, publicEndpoint }),
+    body: JSON.stringify({
+      serverId,
+      publicEndpoint,
+      ...(opts?.disableGreFastPath ? { disableGreFastPath: true } : {}),
+    }),
  })
 }