feat(traffic): классифицировать потоки по брендам и разгрузить ingest

Починить ISO-страну Cloudflare, не держать holder как сервис, клик по срезу открывает Сессии. Не вызывать REST /interface с UDP, flush SQLite пачкой — uptime не должен ловить timeout 10s.

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

backend/package.json

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

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

@@ -157,7 +157,53 @@ try {
   assert.ok(geo.asns?.some((r) => r.label.includes("AS15169")))
   assert.equal(geo.countries?.[0]?.id, "US")
   assert.equal(geo.mapEdges?.[0]?.toCountry, "US")
+  assert.ok(geo.mapEdges?.every((e) => e.toCountry !== "?"))
   assert.equal(geo.conversationsList[0]?.dstCountry, "US")
+  assert.equal(geo.asns?.[0]?.id, "15169")
+} finally {
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+disableRipeEnqueueForTests()
+resetRipeCacheForTests()
+resetFlowCatalogForTests()
+seedRipeCacheForTests({
+  prefix: "1.1.1.0/24",
+  asn: 13335,
+  country: "?",
+  lat: null,
+  lng: null,
+  holder: "CLOUDFLARENET, US",
+  ok: true,
+  fetchedAt: Date.now(),
+})
+rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
+ingestParsedFlowsForServerForTests(7, [
+  {
+    src: "10.1.1.8",
+    dst: "1.1.1.1",
+    proto: 6,
+    srcPort: 51234,
+    dstPort: 443,
+    bytes: 5000,
+    packets: 5,
+    inIface: "2",
+    outIface: "",
+  },
+])
+try {
+  const cf = buildFlowAnalytics({ minutes: 5, serverId: 7 })
+  assert.equal(cf.countries?.[0]?.id, "US")
+  assert.ok(cf.mapEdges?.every((e) => e.toCountry !== "?"))
+  assert.equal(cf.services?.[0]?.label, "Cloudflare")
+  assert.equal(cf.categories?.[0]?.label, "CDN")
+  assert.equal(cf.conversationsList[0]?.dstCountry, "US")
+  assert.equal(cf.asns?.[0]?.id, "13335")
 } finally {
   resetFlowRingsForTests()
   resetIfaceCacheForTests()

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

@@ -14,7 +14,7 @@ import { protoName } from "./traffic-flow-parse.js"
 import {
   getFlowListenerState,
   getRingMbps,
-  listStoredFlowRows,
+  listFlowRowsForWindow,
   type PendingFlowRow,
 } from "./traffic-flow-ingest.js"
 import { resolveIfaceName } from "./traffic-flow-ifaces.js"
@@ -23,6 +23,7 @@ import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
 import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js"
 import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js"
 import { classifyFlowDst, refreshFlowCatalogInBackground } from "./traffic-flow-classify.js"
+import { isIsoCountry } from "./traffic-flow-brands.js"
 
 export interface FlowAnalyticsQuery {
   minutes: number
@@ -37,14 +38,18 @@ function bpsToMbps(bps: number): number {
   return bps / 1_000_000
 }
 
-function topN(map: Map<string, { bytes: number; packets: number }>, windowSec: number, n: number): FlowBreakdownRow[] {
+function topN(
+  map: Map<string, { bytes: number; packets: number; label?: string }>,
+  windowSec: number,
+  n: number,
+): FlowBreakdownRow[] {
   const total = [...map.values()].reduce((a, v) => a + v.bytes, 0) || 1
   return [...map.entries()]
     .sort((a, b) => b[1].bytes - a[1].bytes)
     .slice(0, n)
     .map(([id, v]) => ({
       id,
-      label: id,
+      label: v.label || id,
       bytes: v.bytes,
       packets: v.packets,
       bps: (v.bytes * 8) / windowSec,
@@ -52,10 +57,17 @@ function topN(map: Map<string, { bytes: number; packets: number }>, windowSec: n
     }))
 }
 
-function bump(map: Map<string, { bytes: number; packets: number }>, id: string, bytes: number, packets: number) {
+function bump(
-  const prev = map.get(id) ?? { bytes: 0, packets: 0 }
+  map: Map<string, { bytes: number; packets: number; label?: string }>,
+  id: string,
+  bytes: number,
+  packets: number,
+  label?: string,
+) {
+  const prev = map.get(id) ?? { bytes: 0, packets: 0, label }
   prev.bytes += bytes
   prev.packets += packets
+  if (label) prev.label = label
   map.set(id, prev)
 }
 
@@ -95,13 +107,13 @@ function snapshotStatus(serverId: number): FlowEntityCard["status"] {
   return "online"
 }
 
-function topLabel(map: Map<string, { bytes: number; packets: number }>, fallback = "—"): string {
+function topLabel(map: Map<string, { bytes: number; packets: number; label?: string }>, fallback = "—"): string {
   let best = fallback
   let bestBytes = 0
-  for (const [label, v] of map) {
+  for (const [id, v] of map) {
     if (v.bytes > bestBytes) {
       bestBytes = v.bytes
-      best = label
+      best = v.label || id
     }
   }
   return best
@@ -111,8 +123,7 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
   const settings = getTrafficFlowSettingsRow()
   const top = Math.min(50, Math.max(10, settings.topN))
   const windowSec = Math.max(60, q.minutes * 60)
-  const sinceIso = new Date(Date.now() - q.minutes * 60_000).toISOString()
+  const raw = listFlowRowsForWindow(q.minutes)
-  const raw = listStoredFlowRows(sinceIso)
   const allow = q.userId ? userIfaceAllow(q.userId) : null
   const serverRows = db.select().from(servers).all()
   const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host]))
@@ -122,15 +133,15 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
 
   refreshFlowCatalogInBackground()
 
-  const applications = new Map<string, { bytes: number; packets: number }>()
+  const applications = new Map<string, { bytes: number; packets: number; label?: string }>()
-  const protocols = new Map<string, { bytes: number; packets: number }>()
+  const protocols = new Map<string, { bytes: number; packets: number; label?: string }>()
-  const sources = new Map<string, { bytes: number; packets: number }>()
+  const sources = new Map<string, { bytes: number; packets: number; label?: string }>()
-  const destinations = new Map<string, { bytes: number; packets: number }>()
+  const destinations = new Map<string, { bytes: number; packets: number; label?: string }>()
   const ifacesMap = new Map<string, { bytes: number; packets: number; index: string }>()
-  const asns = new Map<string, { bytes: number; packets: number }>()
+  const asns = new Map<string, { bytes: number; packets: number; label?: string }>()
-  const countries = new Map<string, { bytes: number; packets: number }>()
+  const countries = new Map<string, { bytes: number; packets: number; label?: string }>()
-  const categories = new Map<string, { bytes: number; packets: number }>()
+  const categories = new Map<string, { bytes: number; packets: number; label?: string }>()
-  const services = new Map<string, { bytes: number; packets: number }>()
+  const services = new Map<string, { bytes: number; packets: number; label?: string }>()
   const conv = new Map<string, FlowTalkerDto & { rawBytes: number }>()
   const edgeAcc = new Map<string, FlowMapEdge & { catBytes: Map<string, number> }>()
   const srcs = new Set<string>()
@@ -170,11 +181,13 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
     bump(categories, classified.category, r.bytes, r.packets)
     bump(services, classified.service, r.bytes, r.packets)
     if (ripe?.ok && ripe.asn) {
+      const asnId = String(ripe.asn)
       const asnLabel = ripe.holder ? `AS${ripe.asn} ${ripe.holder}` : `AS${ripe.asn}`
-      bump(asns, asnLabel, r.bytes, r.packets)
+      bump(asns, asnId, r.bytes, r.packets, asnLabel)
     }
-    if (ripe?.ok && ripe.country && ripe.country !== "—") {
+    const dstCountry = ripe?.ok && isIsoCountry(ripe.country) ? ripe.country : ""
-      bump(countries, ripe.country, r.bytes, r.packets)
+    if (dstCountry) {
+      bump(countries, dstCountry, r.bytes, r.packets)
     }
 
     const ckey = wantDedup
@@ -203,13 +216,13 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
         application: app,
         category: classified.category,
         service: classified.service,
-        dstCountry: ripe?.country && ripe.country !== "—" ? ripe.country : undefined,
+        dstCountry: dstCountry || undefined,
         dstAsn: ripe?.asn || undefined,
         rawBytes: r.bytes,
       })
     }
 
-    const toCountry = ripe?.ok && ripe.country && ripe.country !== "—" ? ripe.country : ""
+    const toCountry = dstCountry
     if (toCountry) {
       const fromCountry = countryById.get(r.serverId) || "UN"
       const ekey = `${r.serverId}|${toCountry}`
@@ -348,8 +361,7 @@ function cardFromServer(
 
 export function listFlowExporters(minutes: number): FlowExportersDto {
   const settings = getTrafficFlowSettingsRow()
-  const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString()
+  const rows = listFlowRowsForWindow(minutes)
-  const rows = listStoredFlowRows(sinceIso)
   const ids = new Set<number>()
   for (const r of rows) ids.add(r.serverId)
   for (const p of listHostPeers()) ids.add(p.serverId)

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

@@ -0,0 +1,26 @@
+import assert from "node:assert/strict"
+import {
+  brandByAsn,
+  countryFromHolder,
+  lookupBrand,
+  OTHER_SERVICE,
+  resolveRipeCountry,
+} from "./traffic-flow-brands.js"
+
+assert.equal(resolveRipeCountry("?", 13335, "CLOUDFLARENET, US"), "US")
+assert.equal(resolveRipeCountry("EU", 13335, ""), "US")
+assert.equal(resolveRipeCountry("?", 0, "CLOUDFLARENET, US"), "US")
+assert.equal(countryFromHolder("CLOUDFLARENET, US"), "US")
+assert.equal(resolveRipeCountry("NL", 0, ""), "NL")
+assert.equal(resolveRipeCountry("?", 0, ""), "")
+
+assert.equal(brandByAsn(13335)?.service, "Cloudflare")
+assert.equal(brandByAsn(13335)?.category, "CDN")
+assert.equal(brandByAsn(32590)?.service, "Steam")
+assert.equal(brandByAsn(32590)?.category, "Игры")
+assert.equal(brandByAsn(401115)?.service, "ChatGPT")
+assert.equal(lookupBrand("1.1.1.1", 13335)?.service, "Cloudflare")
+assert.equal(lookupBrand("203.0.113.9", 64500), null)
+assert.equal(OTHER_SERVICE, "Прочее")
+
+console.log("traffic-flow-brands.test.ts: ok")

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

@@ -0,0 +1,105 @@
+import { ipInCidrV4, parseCidrV4 } from "./traffic-flow-ip.js"
+
+export const OTHER_SERVICE = "Прочее"
+
+export interface BrandHit {
+  service: string
+  category: string
+}
+
+const ASN_BRANDS = new Map<number, BrandHit>([
+  [13335, { service: "Cloudflare", category: "CDN" }],
+  [209242, { service: "Cloudflare", category: "CDN" }],
+  [54113, { service: "Fastly", category: "CDN" }],
+  [20940, { service: "Akamai", category: "CDN" }],
+  [16509, { service: "Amazon", category: "CDN" }],
+  [14618, { service: "Amazon", category: "CDN" }],
+  [8075, { service: "Microsoft", category: "CDN" }],
+  [13238, { service: "Yandex", category: "CDN" }],
+  [32590, { service: "Steam", category: "Игры" }],
+  [2906, { service: "Netflix", category: "Видео / стриминг" }],
+  [40027, { service: "Netflix", category: "Видео / стриминг" }],
+  [15169, { service: "Google", category: "Видео / стриминг" }],
+  [36040, { service: "YouTube", category: "Видео / стриминг" }],
+  [46489, { service: "Twitch", category: "Видео / стриминг" }],
+  [401115, { service: "ChatGPT", category: "ИИ" }],
+  [49544, { service: "Discord", category: "Голос" }],
+  [62041, { service: "Telegram", category: "Голос" }],
+  [59930, { service: "Telegram", category: "Голос" }],
+  [211157, { service: "Telegram", category: "Голос" }],
+  [32934, { service: "Meta", category: "CDN" }],
+  [396986, { service: "TikTok", category: "Видео / стриминг" }],
+])
+
+const ASN_HQ_COUNTRY = new Map<number, string>([
+  [13335, "US"],
+  [209242, "US"],
+  [54113, "US"],
+  [20940, "US"],
+  [16509, "US"],
+  [14618, "US"],
+  [8075, "US"],
+  [15169, "US"],
+  [32590, "US"],
+  [2906, "US"],
+  [40027, "US"],
+  [36040, "US"],
+  [46489, "US"],
+  [401115, "US"],
+  [49544, "US"],
+  [32934, "US"],
+  [13238, "RU"],
+  [62041, "NL"],
+  [59930, "NL"],
+  [211157, "NL"],
+])
+
+const CIDR_BRANDS: Array<{ cidr: string; prefixLen: number; hit: BrandHit }> = [
+  { cidr: "104.16.0.0/13", prefixLen: 13, hit: { service: "Cloudflare", category: "CDN" } },
+  { cidr: "104.24.0.0/14", prefixLen: 14, hit: { service: "Cloudflare", category: "CDN" } },
+  { cidr: "172.64.0.0/13", prefixLen: 13, hit: { service: "Cloudflare", category: "CDN" } },
+  { cidr: "162.158.0.0/15", prefixLen: 15, hit: { service: "Cloudflare", category: "CDN" } },
+].sort((a, b) => b.prefixLen - a.prefixLen)
+
+const NON_ISO = new Set(["EU", "AP", "ZZ", "XX", "A1", "A2", "O1"])
+
+export function isIsoCountry(code: string): boolean {
+  const c = String(code ?? "").trim().toUpperCase()
+  return /^[A-Z]{2}$/.test(c) && !NON_ISO.has(c)
+}
+
+export function normalizeIsoCountry(code: string): string {
+  const c = String(code ?? "").trim().toUpperCase()
+  return isIsoCountry(c) ? c : ""
+}
+
+/** `CLOUDFLARENET, US` → `US`. */
+export function countryFromHolder(holder: string): string {
+  const m = String(holder ?? "").trim().match(/,\s*([A-Za-z]{2})\s*$/)
+  return m?.[1] ? normalizeIsoCountry(m[1]) : ""
+}
+
+export function countryForAsn(asn: number): string {
+  if (!asn) return ""
+  return ASN_HQ_COUNTRY.get(asn) ?? ""
+}
+
+export function resolveRipeCountry(country: string, asn: number, holder: string): string {
+  return normalizeIsoCountry(country) || countryFromHolder(holder) || countryForAsn(asn)
+}
+
+export function brandByAsn(asn: number): BrandHit | null {
+  if (!asn) return null
+  return ASN_BRANDS.get(asn) ?? null
+}
+
+export function brandByCidr(ip: string): BrandHit | null {
+  for (const row of CIDR_BRANDS) {
+    if (parseCidrV4(row.cidr) && ipInCidrV4(ip, row.cidr)) return row.hit
+  }
+  return null
+}
+
+export function lookupBrand(ip: string, asn: number): BrandHit | null {
+  return brandByCidr(ip) || brandByAsn(asn)
+}

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

@@ -16,5 +16,10 @@ assert.equal(miss.category, "DNS")
 
 const cdn = classifyFlowDst("203.0.113.9", 6, 443, 1, { prefix: "203.0.113.0/24", asn: 13335, country: "US", lat: null, lng: null, holder: "CLOUDFLARENET", ok: true, fetchedAt: Date.now() })
 assert.equal(cdn.category, "CDN")
+assert.equal(cdn.service, "Cloudflare")
+
+const amazonHolder = classifyFlowDst("203.0.113.50", 6, 443, 1, { prefix: "203.0.113.0/24", asn: 64500, country: "RU", lat: null, lng: null, holder: "AMAZON-AES - Amazon.com, Inc.", ok: true, fetchedAt: Date.now() })
+assert.equal(amazonHolder.service, "Прочее")
+assert.notEqual(amazonHolder.service, "AMAZON-AES - Amazon.com, Inc.")
 
 console.log("traffic-flow-classify.test.ts: ok")

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

@@ -1,3 +1,4 @@
+import { lookupBrand, OTHER_SERVICE } from "./traffic-flow-brands.js"
 import { db } from "../db/index.js"
 import { evobgpSettings } from "../db/schema.js"
 import { ipInCidrV4, parseCidrV4 } from "./traffic-flow-ip.js"
@@ -50,9 +51,10 @@ export function categoryFromPurpose(purpose: string, proto: number, dstPort: num
   if (/streaming|youtube|netflix|twitch|video/.test(p)) return "Видео / стриминг"
   if (/cdn|cloudflare|akamai|fastly/.test(p)) return "CDN"
   if (/voip|discord|zoom/.test(p)) return "Голос"
+  if (/openai|chatgpt|\bai\b/.test(p)) return "ИИ"
   const app = applicationName(proto, dstPort, srcPort)
   if (app === "DNS" || app === "SSH" || app === "BGP") return app
-  return "Проче"
+  return OTHER_SERVICE
 }
 
 function matchCidr(ip: string): CatalogCidr | null {
@@ -70,13 +72,13 @@ export function classifyFlowDst(
   ripe: FlowIpMeta | null,
 ): FlowClassification {
   const hit = matchCidr(dst)
+  const brand = lookupBrand(dst, ripe?.asn ?? 0)
   const asnName = ripe?.asn ? asnPurpose.get(ripe.asn) : undefined
-  const purpose = hit?.purpose || asnName || ripe?.holder || ""
+  const service = (hit?.purpose || brand?.service || asnName || OTHER_SERVICE).trim() || OTHER_SERVICE
-  const service = (hit?.purpose || asnName || ripe?.holder || "Проче").trim() || "Проче"
+  const category = hit
-  return {
+    ? categoryFromPurpose(hit.purpose, proto, dstPort, srcPort)
-    service,
+    : (brand?.category || categoryFromPurpose(asnName || "", proto, dstPort, srcPort))
-    category: categoryFromPurpose(purpose, proto, dstPort, srcPort),
+  return { service, category }
-  }
 }
 
 async function fetchCatalog(): Promise<void> {

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

@@ -4,6 +4,8 @@ import {
   resetIfaceCacheForTests,
   resolveIfaceName,
   rosIdToIfIndex,
+  shouldRefreshIfaces,
+  markIfaceRefreshAttempt,
 } from "./traffic-flow-ifindex.js"
 import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
 
@@ -35,6 +37,14 @@ assert.equal(rosIdToIfIndex("*12"), 18)
 assert.equal(resolveIfaceName(8, "10").name, "gre1")
 assert.equal(resolveIfaceName(8, "18").name, "gre1")
 
+resetIfaceCacheForTests()
+assert.equal(shouldRefreshIfaces(9), true)
+rememberServerIfaces(9, [{ ".id": "*2", name: "ether1" }])
+assert.equal(shouldRefreshIfaces(9), false)
+resetIfaceCacheForTests()
+markIfaceRefreshAttempt(9)
+assert.equal(shouldRefreshIfaces(9), false)
+
 assert.equal(applicationName(6, 443), "HTTPS")
 assert.equal(applicationName(17, 53), "DNS")
 assert.equal(applicationName(6, 22), "SSH")

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

@@ -3,8 +3,9 @@ import { db } from "../db/index.js"
 import { servers } from "../db/schema.js"
 import { MikrotikClient } from "./mikrotik.js"
 import {
-  ifaceCacheFresh,
   rememberServerIfaces,
+  shouldRefreshIfaces,
+  markIfaceRefreshAttempt,
   type RosIfaceIndexRow,
 } from "./traffic-flow-ifindex.js"
 
@@ -15,13 +16,15 @@ export {
   resetIfaceCacheForTests,
   resolveIfaceName,
   rosIdToIfIndex,
+  shouldRefreshIfaces,
+  markIfaceRefreshAttempt,
 } from "./traffic-flow-ifindex.js"
 
 const inflight = new Set<number>()
 
 export async function refreshServerIfaces(serverId: number, force = false): Promise<void> {
   if (inflight.has(serverId)) return
-  if (!force && ifaceCacheFresh(serverId)) return
+  if (!force && !shouldRefreshIfaces(serverId)) return
   inflight.add(serverId)
   try {
     const row = db.select().from(servers).where(eq(servers.id, serverId)).limit(1).all()[0]
@@ -32,6 +35,7 @@ export async function refreshServerIfaces(serverId: number, force = false): Prom
   } catch {
     /* keep previous cache */
   } finally {
+    markIfaceRefreshAttempt(serverId)
     inflight.delete(serverId)
   }
 }

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

@@ -6,6 +6,7 @@ export interface RosIfaceIndexRow {
 
 const cache = new Map<number, Map<number, string>>()
 const fetchedAt = new Map<number, number>()
+const lastAttempt = new Map<number, number>()
 
 export const IFACE_CACHE_TTL_MS = 60_000
 
@@ -52,7 +53,19 @@ export function ifaceCacheFresh(serverId: number, ttlMs = IFACE_CACHE_TTL_MS): b
   return Boolean(prev && Date.now() - prev < ttlMs && cache.has(serverId))
 }
 
+/** Не ходить в REST, пока кэш жив или с момента последней попытки не прошёл TTL. */
+export function shouldRefreshIfaces(serverId: number, ttlMs = IFACE_CACHE_TTL_MS): boolean {
+  if (ifaceCacheFresh(serverId, ttlMs)) return false
+  const attempted = lastAttempt.get(serverId) ?? 0
+  return !(attempted && Date.now() - attempted < ttlMs)
+}
+
+export function markIfaceRefreshAttempt(serverId: number, at = Date.now()): void {
+  lastAttempt.set(serverId, at)
+}
+
 export function resetIfaceCacheForTests(): void {
   cache.clear()
   fetchedAt.clear()
+  lastAttempt.clear()
 }

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

@@ -0,0 +1,44 @@
+import assert from "node:assert/strict"
+import {
+  markIfaceRefreshAttempt,
+  rememberServerIfaces,
+  resetIfaceCacheForTests,
+  shouldRefreshIfaces,
+} from "./traffic-flow-ifindex.js"
+import {
+  lastFlushUsedTransactionForTests,
+  maybeRefreshIfaces,
+  resetFlowRingsForTests,
+  setRefreshIfacesForTests,
+} from "./traffic-flow-ingest.js"
+
+resetIfaceCacheForTests()
+resetFlowRingsForTests()
+
+let refreshCalls = 0
+setRefreshIfacesForTests(async () => {
+  refreshCalls += 1
+})
+
+rememberServerIfaces(1, [{ ".id": "*A", name: "wg-flow" }])
+assert.equal(shouldRefreshIfaces(1), false)
+assert.equal(maybeRefreshIfaces(1), false)
+assert.equal(refreshCalls, 0)
+
+resetIfaceCacheForTests()
+assert.equal(shouldRefreshIfaces(2), true)
+assert.equal(maybeRefreshIfaces(2), true)
+assert.equal(refreshCalls, 1)
+
+markIfaceRefreshAttempt(2)
+assert.equal(shouldRefreshIfaces(2), false)
+assert.equal(maybeRefreshIfaces(2), false)
+assert.equal(refreshCalls, 1)
+
+assert.equal(lastFlushUsedTransactionForTests(), false)
+
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+setRefreshIfacesForTests(null)
+
+console.log("traffic-flow-ingest.test.ts: ok")

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

@@ -1,6 +1,6 @@
 import { createSocket, type Socket } from "node:dgram"
 import { desc, eq, gte, sql } from "drizzle-orm"
-import { db } from "../db/index.js"
+import { db, sqliteDatabase } from "../db/index.js"
 import { flowBuckets, servers } from "../db/schema.js"
 import type { FlowStatsDto, FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
 import { parseFlowPacket, protoName, type ParsedFlow } from "./traffic-flow-parse.js"
@@ -11,7 +11,7 @@ import {
   recordFlowListenerError,
   recordFlowPacket,
 } from "./traffic-flow-settings.js"
-import { ifaceCacheFresh, refreshServerIfaces, resolveIfaceName } from "./traffic-flow-ifaces.js"
+import { refreshServerIfaces, resolveIfaceName, shouldRefreshIfaces } from "./traffic-flow-ifaces.js"
 import { applicationName } from "./traffic-flow-apps.js"
 
 export interface FlowListenerState {
@@ -35,6 +35,8 @@ export interface PendingFlowRow {
 
 const TICK_MS = 2_000
 const RING_LEN = 60
+const LIVE_WINDOW_MS = 15 * 60_000
+const PRUNE_MS = 5 * 60_000
 
 let socket: Socket | null = null
 let state: FlowListenerState = { bound: false, address: null }
@@ -45,7 +47,38 @@ const pending = new Map<string, {
   bytes: number
   packets: number
 }>()
+const recent = new Map<string, PendingFlowRow>()
 let flushTimer: ReturnType<typeof setInterval> | null = null
+let lastPruneAt = 0
+let refreshIfacesImpl: (serverId: number, force?: boolean) => Promise<void> = refreshServerIfaces
+let lastFlushUsedTransaction = false
+
+const upsertFlowStmt = sqliteDatabase.prepare(`
+  INSERT INTO flow_buckets (
+    server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface
+  ) VALUES (
+    @serverId, @bucketAt, @src, @dst, @proto, @srcPort, @dstPort, @bytes, @packets, @inIface
+  )
+  ON CONFLICT(server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
+  DO UPDATE SET
+    bytes = bytes + excluded.bytes,
+    packets = packets + excluded.packets
+`)
+
+const upsertFlowTx = sqliteDatabase.transaction((rows: Array<{
+  serverId: number
+  bucketAt: string
+  src: string
+  dst: string
+  proto: number
+  srcPort: number
+  dstPort: number
+  bytes: number
+  packets: number
+  inIface: string
+}>) => {
+  for (const row of rows) upsertFlowStmt.run(row)
+})
 
 const tickAccum = new Map<string, { inBytes: number; outBytes: number }>()
 const rings = new Map<string, { inBps: number[]; outBps: number[] }>()
@@ -142,12 +175,55 @@ function resolveServerId(exporterIp: string): number | null {
   })
 }
 
+export function setRefreshIfacesForTests(fn: typeof refreshServerIfaces | null): void {
+  refreshIfacesImpl = fn ?? refreshServerIfaces
+}
+
+/** REST /interface только при протухшем TTL, не из-за #N в пакете. */
+export function maybeRefreshIfaces(serverId: number): boolean {
+  if (!shouldRefreshIfaces(serverId)) return false
+  void refreshIfacesImpl(serverId)
+  return true
+}
+
+export function lastFlushUsedTransactionForTests(): boolean {
+  return lastFlushUsedTransaction
+}
+
+function pendingKey(serverId: number, bucketAt: string, flow: ParsedFlow): string {
+  return `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}\0${flow.inIface}`
+}
+
+function rowKey(row: PendingFlowRow): string {
+  return `${row.serverId}|${row.bucketAt}|${row.src}|${row.dst}|${row.proto}|${row.srcPort}|${row.dstPort}|${row.inIface}`
+}
+
+function mergeInto(map: Map<string, PendingFlowRow>, row: PendingFlowRow): void {
+  const key = rowKey(row)
+  const prev = map.get(key)
+  if (prev) {
+    prev.bytes += row.bytes
+    prev.packets += row.packets
+    return
+  }
+  map.set(key, { ...row })
+}
+
+function rememberRecent(rows: PendingFlowRow[]): void {
+  for (const row of rows) mergeInto(recent, row)
+}
+
+function pruneRecent(sinceMs = Date.now() - LIVE_WINDOW_MS): void {
+  const cutoff = new Date(sinceMs).toISOString()
+  for (const [key, row] of recent) {
+    if (row.bucketAt < cutoff) recent.delete(key)
+  }
+}
+
 function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean {
   const serverId = resolveServerId(exporterIp)
   if (serverId == null) return false
-  const needsRefresh = !ifaceCacheFresh(serverId)
+  maybeRefreshIfaces(serverId)
-    || flows.some((f) => resolveIfaceName(serverId, f.inIface).name.startsWith("#"))
-  if (needsRefresh) void refreshServerIfaces(serverId, true)
   const bucketAt = minuteBucketIso()
   for (const flow of flows) {
     addToTick(serverId, flow.inIface, flow.outIface, flow.bytes)
@@ -170,7 +246,17 @@ function queueFlows(exporterIp: string, flows: ParsedFlow[]): boolean {
 }
 
 export function peekPendingFlows(): PendingFlowRow[] {
-  return [...pending.values()].map((row) => ({
+  return [...pending.values()].map(toPendingRow)
+}
+
+function toPendingRow(row: {
+  serverId: number
+  bucketAt: string
+  flow: ParsedFlow
+  bytes: number
+  packets: number
+}): PendingFlowRow {
+  return {
     serverId: row.serverId,
     bucketAt: row.bucketAt,
     src: row.flow.src || "0.0.0.0",
@@ -182,53 +268,17 @@ export function peekPendingFlows(): PendingFlowRow[] {
     packets: row.packets,
     inIface: row.flow.inIface,
     outIface: row.flow.outIface,
-  }))
+  }
 }
 
-function flushPending() {
+function pruneStoredBuckets(): void {
-  if (pending.size === 0) return
+  const now = Date.now()
+  if (now - lastPruneAt < PRUNE_MS) return
+  lastPruneAt = now
   const settings = getTrafficFlowSettingsRow()
   const topN = Math.max(20, settings.topN)
-  const cutoff = new Date(Date.now() - settings.retentionHours * 3600_000).toISOString()
+  const cutoff = new Date(now - settings.retentionHours * 3600_000).toISOString()
-  const rows = [...pending.values()]
-  pending.clear()
-
-  for (const row of rows) {
-    try {
-      db.insert(flowBuckets).values({
-        serverId: row.serverId,
-        bucketAt: row.bucketAt,
-        src: row.flow.src || "0.0.0.0",
-        dst: row.flow.dst || "0.0.0.0",
-        proto: row.flow.proto,
-        srcPort: row.flow.srcPort,
-        dstPort: row.flow.dstPort,
-        bytes: row.bytes,
-        packets: row.packets,
-        inIface: row.flow.inIface,
-      }).onConflictDoUpdate({
-        target: [
-          flowBuckets.serverId,
-          flowBuckets.bucketAt,
-          flowBuckets.src,
-          flowBuckets.dst,
-          flowBuckets.proto,
-          flowBuckets.srcPort,
-          flowBuckets.dstPort,
-          flowBuckets.inIface,
-        ],
-        set: {
-          bytes: sql`${flowBuckets.bytes} + excluded.bytes`,
-          packets: sql`${flowBuckets.packets} + excluded.packets`,
-        },
-      }).run()
-    } catch {
-      // ignore single-row failures
-    }
-  }
-
   db.delete(flowBuckets).where(sql`${flowBuckets.bucketAt} < ${cutoff}`).run()
-
   const latest = db.select({ bucketAt: flowBuckets.bucketAt }).from(flowBuckets)
     .orderBy(desc(flowBuckets.bucketAt)).limit(1).all()[0]?.bucketAt
   if (!latest) return
@@ -248,6 +298,58 @@ function flushPending() {
   }
 }
 
+function flushPending() {
+  pruneRecent()
+  if (pending.size === 0) {
+    pruneStoredBuckets()
+    lastFlushUsedTransaction = false
+    return
+  }
+  const rows = [...pending.values()].map(toPendingRow)
+  pending.clear()
+  rememberRecent(rows)
+  lastFlushUsedTransaction = false
+  try {
+    upsertFlowTx(rows.map((r) => ({
+      serverId: r.serverId,
+      bucketAt: r.bucketAt,
+      src: r.src,
+      dst: r.dst,
+      proto: r.proto,
+      srcPort: r.srcPort,
+      dstPort: r.dstPort,
+      bytes: r.bytes,
+      packets: r.packets,
+      inIface: r.inIface,
+    })))
+    lastFlushUsedTransaction = true
+  } catch {
+    for (const r of rows) {
+      try {
+        upsertFlowStmt.run({
+          serverId: r.serverId,
+          bucketAt: r.bucketAt,
+          src: r.src,
+          dst: r.dst,
+          proto: r.proto,
+          srcPort: r.srcPort,
+          dstPort: r.dstPort,
+          bytes: r.bytes,
+          packets: r.packets,
+          inIface: r.inIface,
+        })
+      } catch {
+        /* ignore single-row failures */
+      }
+    }
+  }
+  pruneStoredBuckets()
+}
+
+export function flushPendingForTests(): void {
+  flushPending()
+}
+
 function onTick() {
   rollFlowRings()
   flushPending()
@@ -306,12 +408,24 @@ export function startTrafficFlowListener() {
   flushTimer = setInterval(onTick, TICK_MS)
 }
 
+export function listLiveFlowRows(sinceIso: string): PendingFlowRow[] {
+  const merged = new Map<string, PendingFlowRow>()
+  for (const row of recent.values()) {
+    if (row.bucketAt < sinceIso) continue
+    mergeInto(merged, row)
+  }
+  for (const row of peekPendingFlows()) {
+    if (row.bucketAt < sinceIso) continue
+    mergeInto(merged, row)
+  }
+  return [...merged.values()]
+}
+
 export function listStoredFlowRows(sinceIso: string): PendingFlowRow[] {
   const stored = db.select().from(flowBuckets).where(gte(flowBuckets.bucketAt, sinceIso)).all()
   const merged = new Map<string, PendingFlowRow>()
   for (const r of stored) {
-    const key = `${r.serverId}|${r.bucketAt}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}|${r.inIface}`
+    mergeInto(merged, {
-    merged.set(key, {
       serverId: r.serverId,
       bucketAt: r.bucketAt,
       src: r.src,
@@ -327,22 +441,21 @@ export function listStoredFlowRows(sinceIso: string): PendingFlowRow[] {
   }
   for (const p of peekPendingFlows()) {
     if (p.bucketAt < sinceIso) continue
-    const key = `${p.serverId}|${p.bucketAt}|${p.src}|${p.dst}|${p.proto}|${p.srcPort}|${p.dstPort}|${p.inIface}`
+    mergeInto(merged, p)
-    const prev = merged.get(key)
-    if (prev) {
-      prev.bytes += p.bytes
-      prev.packets += p.packets
-    } else {
-      merged.set(key, { ...p })
-    }
   }
   return [...merged.values()]
 }
 
+/** SSE / короткое окно — память; длинные окна — SQLite. */
+export function listFlowRowsForWindow(minutes: number): PendingFlowRow[] {
+  const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString()
+  if (minutes <= 15) return listLiveFlowRows(sinceIso)
+  return listStoredFlowRows(sinceIso)
+}
+
 export function listFlowTalkers(minutes = 5): FlowStatsDto {
   const settings = getTrafficFlowSettingsRow()
-  const rangeStart = new Date(Date.now() - minutes * 60_000).toISOString()
+  const rows = listFlowRowsForWindow(minutes)
-  const rows = listStoredFlowRows(rangeStart)
   const serverRows = db.select().from(servers).all()
   const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host]))
   const agg = new Map<string, FlowTalkerDto & { rawBytes: number }>()
@@ -426,7 +539,7 @@ export function ingestParsedFlowsForServerForTests(serverId: number, flows: Pars
   const bucketAt = minuteBucketIso()
   for (const flow of flows) {
     addToTick(serverId, flow.inIface, flow.outIface, flow.bytes)
-    const key = `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}\0${flow.inIface}`
+    const key = pendingKey(serverId, bucketAt, flow)
     const prev = pending.get(key)
     if (prev) {
       prev.bytes += flow.bytes
@@ -448,4 +561,8 @@ export function resetFlowRingsForTests() {
   tickAccum.clear()
   rings.clear()
   pending.clear()
+  recent.clear()
+  lastPruneAt = 0
+  lastFlushUsedTransaction = false
+  refreshIfacesImpl = refreshServerIfaces
 }

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

@@ -69,4 +69,35 @@ enqueueRipeMisses(["203.0.113.50"])
 await flushRipeQueueForTests()
 assert.equal(ripeFetchCountForTests(), afterNeg)
 
+resetRipeCacheForTests()
+disableRipePersistForTests()
+seedRipeCacheForTests({
+  prefix: "1.1.1.0/24",
+  asn: 13335,
+  country: "?",
+  lat: null,
+  lng: null,
+  holder: "CLOUDFLARENET, US",
+  ok: true,
+  fetchedAt: Date.now(),
+})
+assert.equal(lookupRipeCached("1.1.1.1")?.country, "US")
+assert.ok(lookupRipeCached("1.1.1.1")?.country !== "?")
+
+resetRipeCacheForTests()
+disableRipePersistForTests()
+setRipeFetchForTests(async (input) => {
+  const url = String(input)
+  const body = url.includes("network-info")
+    ? { data: { prefix: "1.0.0.0/24", asns: ["13335"] } }
+    : url.includes("maxmind-geo-lite")
+      ? { data: { located_resources: [{ locations: [{ country: "?" }] }] } }
+      : { data: { holder: "CLOUDFLARENET, US" } }
+  return new Response(JSON.stringify(body), { status: 200, headers: { "Content-Type": "application/json" } })
+})
+enqueueRipeMisses(["1.0.0.1"])
+await flushRipeQueueForTests()
+assert.equal(lookupRipeCached("1.0.0.1")?.country, "US")
+assert.equal(lookupRipeCached("1.0.0.1")?.asn, 13335)
+
 console.log("traffic-flow-ripe.test.ts: ok")

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

@@ -1,5 +1,6 @@
 import { sqliteDatabase } from "../db/index.js"
 import { ipInCidrV4, ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
+import { resolveRipeCountry } from "./traffic-flow-brands.js"
 
 export interface FlowIpMeta {
   prefix: string
@@ -15,6 +16,7 @@ export interface FlowIpMeta {
 const HIT_TTL_MS = 24 * 60 * 60_000
 const NEG_TTL_MS = 6 * 60 * 60_000
 const MAX_NEW_PREFIX_PER_MIN = 30
+const MAX_QUEUE = 90
 const CONCURRENCY = 3
 const RIPE_BASE = "https://stat.ripe.net/data"
 const UA = "MikrotikManager-flow/1.0"
@@ -107,13 +109,15 @@ function loadSqlite(): void {
     }>
     for (const r of rows) {
       const fetchedAt = Date.parse(r.fetched_at)
+      const asn = Number(r.asn ?? 0) || 0
+      const holder = r.holder || ""
       mem.set(r.prefix, {
         prefix: r.prefix,
-        asn: Number(r.asn ?? 0) || 0,
+        asn,
-        country: r.country || "—",
+        country: resolveRipeCountry(r.country || "", asn, holder) || "—",
         lat: r.lat == null ? null : Number(r.lat),
         lng: r.lng == null ? null : Number(r.lng),
-        holder: r.holder || "",
+        holder,
         ok: r.ok !== 0,
         fetchedAt: Number.isFinite(fetchedAt) ? fetchedAt : 0,
       })
@@ -206,6 +210,8 @@ export function lookupRipeCached(ip: string): FlowIpMeta | null {
     }
   }
   return best
+    ? { ...best, country: resolveRipeCountry(best.country, best.asn, best.holder) || "—" }
+    : null
 }
 
 async function ripeJson(path: string, resource: string): Promise<unknown> {
@@ -247,7 +253,7 @@ function pickGeo(data: unknown): { country: string; lat: number | null; lng: num
     }
   }
   const loc = d?.data?.located_resources?.[0]?.locations?.[0]
-  const country = String(loc?.country ?? "").trim().toUpperCase()
+  const country = resolveRipeCountry(String(loc?.country ?? ""), 0, "")
   const lat = loc?.latitude == null ? null : Number(loc.latitude)
   const lng = loc?.longitude == null ? null : Number(loc.longitude)
   return {
@@ -299,14 +305,15 @@ async function resolveIp(ip: string): Promise<FlowIpMeta | null> {
           /* best-effort */
         }
       }
+      const country = resolveRipeCountry(geo.country, asn, holder)
       const entry: FlowIpMeta = {
         prefix,
         asn,
-        country: geo.country,
+        country: country || "—",
         lat: geo.lat,
         lng: geo.lng,
         holder,
-        ok: Boolean(asn || (geo.country && geo.country !== "—")),
+        ok: Boolean(asn || country),
         fetchedAt: Date.now(),
       }
       mem.set(prefix, entry)
@@ -360,6 +367,7 @@ export function enqueueRipeMisses(ips: Iterable<string>): void {
   if (!enqueueEnabled) return
   loadSqlite()
   for (const raw of ips) {
+    if (queue.length >= MAX_QUEUE) break
     const ip = String(raw ?? "").trim()
     if (!ip || isNonPublicIp(ip)) continue
     if (lookupRipeCached(ip)) continue

components/flag.tsx

@@ -46,6 +46,7 @@ function nearestCdnSize(px: number): number {
 export function Flag({ code, size = 20, className }: FlagProps) {
   if (!code) return null
   const lower  = code.toLowerCase()
+  if (!/^[a-z]{2}$/.test(lower)) return null
   const name   = countryName(code.toUpperCase())
   const cdnSrc = nearestCdnSize(size)
   const cdnSrc2x = nearestCdnSize(size * 2)

components/traffic/flow-analytics-panel.tsx

@@ -2,7 +2,7 @@
 
 import { useMemo, useState } from "react"
 import { type ColumnDef, getCoreRowModel, useReactTable } from "@tanstack/react-table"
-import type { FlowAnalyticsDto, FlowBreakdownRow, FlowEntityCard } from "@mmapp/contracts/traffic-flow"
+import type { FlowAnalyticsDto, FlowBreakdownRow, FlowEntityCard, FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
 import { ArrowDownIcon, ArrowUpIcon, GitBranchIcon, GlobeIcon, LayersIcon, UsersIcon } from "lucide-react"
 import { Badge } from "@/components/reui/badge"
 import { KpiStatGrid } from "@/components/reui-kit/kpi-stat-grid"
@@ -108,14 +108,36 @@ export function FlowEntityCardView({
   )
 }
 
+type SessionFilter = {
+  kind: "application" | "category" | "service" | "asn" | "country" | "protocol" | "source" | "destination" | "iface"
+  value: string
+  label: string
+}
+
+function talkerMatchesFilter(row: FlowTalkerDto, filter: SessionFilter): boolean {
+  switch (filter.kind) {
+    case "application": return row.application === filter.value
+    case "category": return row.category === filter.value
+    case "service": return row.service === filter.value
+    case "asn": return String(row.dstAsn ?? "") === filter.value
+    case "country": return row.dstCountry === filter.value
+    case "protocol": return row.protoName === filter.value
+    case "source": return row.src === filter.value
+    case "destination": return row.dst === filter.value
+    case "iface": return row.inIface === filter.value
+  }
+}
+
 function FlowBreakdownGrid({
   rows,
   empty,
   country,
+  onPick,
 }: {
   rows: FlowBreakdownRow[]
   empty?: string
   country?: boolean
+  onPick?: (row: FlowBreakdownRow) => void
 }) {
   const columns = useMemo<ColumnDef<FlowBreakdownRow>[]>(
     () => [
@@ -178,7 +200,12 @@ function FlowBreakdownGrid({
   })
 
   return (
-    <DataGridShell table={table} recordCount={rows.length} emptyMessage={empty ?? "Нет данных за период"} />
+    <DataGridShell
+      table={table}
+      recordCount={rows.length}
+      emptyMessage={empty ?? "Нет данных за период"}
+      onRowClick={onPick}
+    />
   )
 }
 
@@ -206,13 +233,18 @@ export function FlowAnalyticsDetail({
   emptyHint?: string
 }) {
   const [slice, setSlice] = useState("applications")
-  const [mapCountry, setMapCountry] = useState<string | null>(null)
+  const [sessionFilter, setSessionFilter] = useState<SessionFilter | null>(null)
   const rxNow = analytics ? analytics.bpsNow / 1_000_000 : (card?.rxNow ?? 0)
   const bytes = analytics?.bytes ?? card?.bytes ?? 0
   const sessionRows = (analytics?.conversationsList ?? []).filter((row) =>
-    mapCountry ? row.dstCountry === mapCountry : true,
+    sessionFilter ? talkerMatchesFilter(row, sessionFilter) : true,
   )
 
+  function pickBreakdown(kind: SessionFilter["kind"], row: FlowBreakdownRow) {
+    setSessionFilter({ kind, value: row.id, label: row.label })
+    setSlice("sessions")
+  }
+
   if (!card) {
     return (
       <p className="text-sm text-muted-foreground py-8 text-center">
@@ -375,47 +407,47 @@ export function FlowAnalyticsDetail({
             <TabsTrigger value="interfaces">Интерфейсы</TabsTrigger>
           </TabsList>
           <TabsContent value="applications">
-            <FlowBreakdownGrid rows={analytics?.applications ?? []} />
+            <FlowBreakdownGrid rows={analytics?.applications ?? []} onPick={(row) => pickBreakdown("application", row)} />
           </TabsContent>
           <TabsContent value="categories">
-            <FlowBreakdownGrid rows={analytics?.categories ?? []} />
+            <FlowBreakdownGrid rows={analytics?.categories ?? []} onPick={(row) => pickBreakdown("category", row)} />
           </TabsContent>
           <TabsContent value="services">
-            <FlowBreakdownGrid rows={analytics?.services ?? []} />
+            <FlowBreakdownGrid rows={analytics?.services ?? []} onPick={(row) => pickBreakdown("service", row)} />
           </TabsContent>
           <TabsContent value="asns">
-            <FlowBreakdownGrid rows={analytics?.asns ?? []} />
+            <FlowBreakdownGrid rows={analytics?.asns ?? []} onPick={(row) => pickBreakdown("asn", row)} />
           </TabsContent>
           <TabsContent value="countries">
-            <FlowBreakdownGrid rows={analytics?.countries ?? []} country />
+            <FlowBreakdownGrid rows={analytics?.countries ?? []} country onPick={(row) => pickBreakdown("country", row)} />
           </TabsContent>
           <TabsContent value="map">
             <FlowTrafficMap
               edges={analytics?.mapEdges ?? []}
               onSelectCountry={(iso) => {
-                setMapCountry(iso)
+                setSessionFilter({ kind: "country", value: iso, label: iso })
                 setSlice("sessions")
               }}
             />
           </TabsContent>
           <TabsContent value="protocols">
-            <FlowBreakdownGrid rows={analytics?.protocols ?? []} />
+            <FlowBreakdownGrid rows={analytics?.protocols ?? []} onPick={(row) => pickBreakdown("protocol", row)} />
           </TabsContent>
           <TabsContent value="sources">
-            <FlowBreakdownGrid rows={analytics?.sources ?? []} />
+            <FlowBreakdownGrid rows={analytics?.sources ?? []} onPick={(row) => pickBreakdown("source", row)} />
           </TabsContent>
           <TabsContent value="destinations">
-            <FlowBreakdownGrid rows={analytics?.destinations ?? []} />
+            <FlowBreakdownGrid rows={analytics?.destinations ?? []} onPick={(row) => pickBreakdown("destination", row)} />
           </TabsContent>
           <TabsContent value="sessions">
-            {mapCountry ? (
+            {sessionFilter ? (
               <div className="flex items-center gap-2 mb-2">
                 <GlobeIcon className="size-3.5 text-muted-foreground" />
-                <span className="text-xs text-muted-foreground">фильтр страны {mapCountry}</span>
+                <Badge variant="secondary" size="sm">Фильтр: {sessionFilter.label}</Badge>
                 <button
                   type="button"
                   className="text-xs text-primary"
-                  onClick={() => setMapCountry(null)}
+                  onClick={() => setSessionFilter(null)}
                 >
                   сбросить
                 </button>
@@ -423,13 +455,14 @@ export function FlowAnalyticsDetail({
             ) : null}
             <TrafficFlowsDataGrid
               rows={sessionRows}
-              emptyHint={emptyHint}
+              emptyHint={emptyHint ?? "Нет сессий по выбранному фильтру"}
             />
           </TabsContent>
           <TabsContent value="interfaces">
             <FlowBreakdownGrid
               rows={analytics?.interfaces ?? []}
               empty="Нет данных по интерфейсам"
+              onPick={(row) => pickBreakdown("iface", row)}
             />
           </TabsContent>
         </Tabs>

components/traffic/flow-traffic-map.tsx

@@ -72,7 +72,9 @@ function FlowTrafficMap({
         header: () => <span className="text-xs font-medium text-muted-foreground">Назначение</span>,
         cell: ({ row }) => (
           <span className="flex items-center gap-1.5 text-sm">
-            <Flag code={row.original.toCountry} />
+            {/^[a-z]{2}$/i.test(row.original.toCountry)
+              ? <Flag code={row.original.toCountry} />
+              : null}
             {row.original.toCountry}
             {row.original.toAsn ? <span className="font-mono text-[10px] text-muted-foreground">AS{row.original.toAsn}</span> : null}
           </span>
@@ -176,7 +178,12 @@ function FlowTrafficMap({
           ) : null}
         </FramePanel>
       </Frame>
-      <DataGridShell table={table} recordCount={edges.length} emptyMessage="Нет рёбер с известной страной" />
+      <DataGridShell
+        table={table}
+        recordCount={edges.length}
+        emptyMessage="Нет рёбер с известной страной"
+        onRowClick={(row) => onSelectCountry?.(row.toCountry)}
+      />
     </div>
   )
 }