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

Починить ISO-страну Cloudflare, не держать holder как сервис, клик по срезу открывает Сессии. Не вызывать REST /interface с UDP, flush SQLite пачкой — uptime не должен ловить timeout 10s. Co-authored-by: Cursor <[email protected]>
feat(traffic): добавить карту и срезы IPFIX по странам и сессиям
2026-09-07 03:01:14 +07:00 · 2026-09-07 02:13:21 +07:00
30 changed files with 1908 additions and 168 deletions
@@ -67,7 +67,7 @@ function flowEmptyHint(stats: FlowStatsDto | null): string | undefined {
  if (!stats) return undefined
  if (stats.lastError) return stats.lastError
  if (stats.packetsReceived) {
-    return `IPFIX приходит (${stats.lastExporterIp ?? "экспортёр"}), но разговоры ещё не записаны.`
+    return `IPFIX приходит (${stats.lastExporterIp ?? "экспортёр"}), но сессии ещё не записаны.`
  }
  if (stats.listenerBound === false) {
    return "Коллектор UDP не слушает. Подключите JH ещё раз — ingest включится автоматически."
@@ -771,6 +771,7 @@ export default function TrafficPage() {
  const [flowClients, setFlowClients] = useState<FlowEntityCard[]>([])
  const [flowAnalytics, setFlowAnalytics] = useState<FlowAnalyticsDto | null>(null)
  const [flowIface, setFlowIface] = useState("__all__")
+  const [flowDedup, setFlowDedup] = useState(true)
  const [overlayOpen, setOverlayOpen] = useState(false)
  const [catalogServers, setCatalogServers] = useState<ServerRead[]>([])
  const effectiveMode: GroupMode = groupMode
@@ -788,6 +789,7 @@ export default function TrafficPage() {
    serverId: flowScope === "servers" ? selectedId : undefined,
    userId: flowScope === "users" ? selectedId : undefined,
    iface: flowIface,
+    dedup: flowDedup,
  })

  const toLiveServer = (s: LiveTrafficServer): ServerTraffic => {
@@ -914,8 +916,9 @@ export default function TrafficPage() {
      serverId: flowScope === "servers" ? selectedId : undefined,
      userId: flowScope === "users" ? selectedId : undefined,
      iface: flowIface,
+      dedup: flowDedup,
    }).then(setFlowAnalytics).catch(() => setFlowAnalytics(null))
-  }, [isLive, effectiveMode, selectedId, range, flowScope, flowIface, backendUrl])
+  }, [isLive, effectiveMode, selectedId, range, flowScope, flowIface, flowDedup, backendUrl])

  useEffect(() => {
    setFlowIface("__all__")
@@ -1269,6 +1272,8 @@ export default function TrafficPage() {
                      onRange={(r) => setRange(r as Range)}
                      selectedIface={flowIface}
                      onIface={setFlowIface}
+                      dedup={flowDedup}
+                      onDedup={setFlowDedup}
                      liveHint={displayedFlow?.live ? "live" : undefined}
                      emptyHint={flowEmptyHint(flowStats)}
                    />
@@ -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-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": {
@@ -162,6 +162,22 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_flow_buckets_unique
 CREATE INDEX IF NOT EXISTS idx_flow_buckets_server_time
  ON flow_buckets(server_id, bucket_at);

+CREATE TABLE IF NOT EXISTS flow_ip_meta (
+  prefix TEXT PRIMARY KEY,
+  asn INTEGER NOT NULL DEFAULT 0,
+  country TEXT NOT NULL DEFAULT '',
+  lat REAL,
+  lng REAL,
+  holder TEXT NOT NULL DEFAULT '',
+  ok INTEGER NOT NULL DEFAULT 1,
+  fetched_at TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS flow_asn_meta (
+  asn INTEGER PRIMARY KEY,
+  holder TEXT NOT NULL DEFAULT '',
+  fetched_at TEXT NOT NULL
+);
+
 CREATE TABLE IF NOT EXISTS uptime_settings (
  id INTEGER PRIMARY KEY,
  enabled INTEGER NOT NULL DEFAULT 1,
@@ -202,6 +202,23 @@ export const flowBuckets = sqliteTable("flow_buckets", {
  ),
 ])

+export const flowIpMeta = sqliteTable("flow_ip_meta", {
+  prefix:    text("prefix").primaryKey(),
+  asn:       integer("asn").notNull().default(0),
+  country:   text("country").notNull().default(""),
+  lat:       real("lat"),
+  lng:       real("lng"),
+  holder:    text("holder").notNull().default(""),
+  ok:        integer("ok").notNull().default(1),
+  fetchedAt: text("fetched_at").notNull(),
+})
+
+export const flowAsnMeta = sqliteTable("flow_asn_meta", {
+  asn:       integer("asn").primaryKey(),
+  holder:    text("holder").notNull().default(""),
+  fetchedAt: text("fetched_at").notNull(),
+})
+
 export const trafficSamples = sqliteTable("traffic_samples", {
  id:            integer("id").primaryKey({ autoIncrement: true }),
  serverId:      integer("server_id")
@@ -641,6 +658,8 @@ export type RecursiveRouteRow = typeof recursiveRoutes.$inferSelect
 export type TrafficSettingsRow = typeof trafficSettings.$inferSelect
 export type TrafficFlowSettingsRow = typeof trafficFlowSettings.$inferSelect
 export type FlowBucketRow = typeof flowBuckets.$inferSelect
+export type FlowIpMetaRow = typeof flowIpMeta.$inferSelect
+export type FlowAsnMetaRow = typeof flowAsnMeta.$inferSelect
 export type ServersApiPingSettingsRow = typeof serversApiPingSettings.$inferSelect
 export type TrafficSampleRow = typeof trafficSamples.$inferSelect
 export type UptimeSettingsRow = typeof uptimeSettings.$inferSelect
@@ -43,13 +43,20 @@ function parseId(raw: unknown): number | undefined {
  return Number.isFinite(n) ? n : undefined
 }

+function parseDedup(raw: unknown): boolean {
+  if (raw == null || raw === "") return true
+  const s = String(raw).toLowerCase()
+  return s !== "0" && s !== "false" && s !== "off"
+}
+
 function analyticsQuery(req: FastifyRequest) {
-  const q = req.query as { range?: string; serverId?: string; userId?: string; iface?: string }
+  const q = req.query as { range?: string; serverId?: string; userId?: string; iface?: string; dedup?: string }
  return {
    minutes: rangeToMinutes(q.range),
    serverId: parseId(q.serverId),
    userId: q.userId?.trim() || undefined,
    iface: q.iface?.trim() || undefined,
+    dedup: parseDedup(q.dedup),
  }
 }

@@ -5,7 +5,19 @@ import {
  resetFlowRingsForTests,
 } from "./traffic-flow-ingest.js"
 import { buildFlowAnalytics } from "./traffic-flow-analytics.js"
+import { disableCatalogFetchForTests, resetFlowCatalogForTests, seedFlowCatalogForTests } from "./traffic-flow-classify.js"
+import {
+  disableRipeEnqueueForTests,
+  disableRipePersistForTests,
+  resetRipeCacheForTests,
+  seedRipeCacheForTests,
+} from "./traffic-flow-ripe.js"

+disableCatalogFetchForTests()
+resetFlowCatalogForTests()
+disableRipePersistForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
 resetIfaceCacheForTests()
 resetFlowRingsForTests()
 rememberServerIfaces(7, [
@@ -65,4 +77,138 @@ try {
  resetIfaceCacheForTests()
 }

+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "ether1" },
+  { ".id": "*A", name: "wg-flow" },
+])
+ingestParsedFlowsForServerForTests(7, [
+  {
+    src: "10.1.1.8",
+    dst: "8.8.8.8",
+    proto: 6,
+    srcPort: 51234,
+    dstPort: 443,
+    bytes: 12_000,
+    packets: 10,
+    inIface: "2",
+    outIface: "10",
+  },
+  {
+    src: "10.1.1.8",
+    dst: "8.8.8.8",
+    proto: 6,
+    srcPort: 51234,
+    dstPort: 443,
+    bytes: 9_000,
+    packets: 9,
+    inIface: "10",
+    outIface: "",
+  },
+])
+try {
+  const summed = buildFlowAnalytics({ minutes: 5, serverId: 7, dedup: false })
+  assert.equal(summed.bytes, 21_000)
+  assert.equal(summed.conversations, 2)
+  const deduped = buildFlowAnalytics({ minutes: 5, serverId: 7, dedup: true })
+  assert.equal(deduped.bytes, 12_000)
+  assert.equal(deduped.conversations, 1)
+  assert.equal(deduped.dedupApplied, true)
+  assert.equal(deduped.interfaces.length, 2)
+} finally {
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+}
+
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+disableRipeEnqueueForTests()
+seedRipeCacheForTests({
+  prefix: "8.8.8.0/24",
+  asn: 15169,
+  country: "US",
+  lat: 37.4,
+  lng: -122.1,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+})
+seedFlowCatalogForTests({
+  cidrs: [{ cidr: "8.8.8.0/24", purpose: "steam-gaming" }],
+})
+rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
+ingestParsedFlowsForServerForTests(7, [
+  {
+    src: "10.1.1.8",
+    dst: "8.8.8.8",
+    proto: 6,
+    srcPort: 51234,
+    dstPort: 443,
+    bytes: 12_000,
+    packets: 10,
+    inIface: "2",
+    outIface: "",
+  },
+])
+try {
+  const geo = buildFlowAnalytics({ minutes: 5, serverId: 7 })
+  assert.equal(geo.categories?.[0]?.label, "Игры")
+  assert.ok(geo.asns?.some((r) => r.label.includes("AS15169")))
+  assert.equal(geo.countries?.[0]?.id, "US")
+  assert.equal(geo.mapEdges?.[0]?.toCountry, "US")
+  assert.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()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
 console.log("traffic-flow-analytics.test.ts: ok")
@@ -7,38 +7,49 @@ import type {
  FlowClientsDto,
  FlowEntityCard,
  FlowExportersDto,
+  FlowMapEdge,
  FlowTalkerDto,
 } from "@mmapp/contracts/traffic-flow"
 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"
 import { getTrafficFlowSettingsRow, listHostPeers } from "./traffic-flow-settings.js"
 import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
+import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js"
+import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js"
+import { classifyFlowDst, refreshFlowCatalogInBackground } from "./traffic-flow-classify.js"
+import { isIsoCountry } from "./traffic-flow-brands.js"

 export interface FlowAnalyticsQuery {
  minutes: number
  serverId?: number
  userId?: string
  iface?: string
+  /** Default true: один 5-tuple = max байт по ifaces. */
+  dedup?: boolean
 }

 function bpsToMbps(bps: number): number {
  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,
@@ -46,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) {
-  const prev = map.get(id) ?? { bytes: 0, packets: 0 }
+function bump(
+  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)
 }

@@ -89,48 +107,92 @@ function snapshotStatus(serverId: number): FlowEntityCard["status"] {
  return "online"
 }

+function topLabel(map: Map<string, { bytes: number; packets: number; label?: string }>, fallback = "—"): string {
+  let best = fallback
+  let bestBytes = 0
+  for (const [id, v] of map) {
+    if (v.bytes > bestBytes) {
+      bestBytes = v.bytes
+      best = v.label || id
+    }
+  }
+  return best
+}
+
 export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
  const settings = getTrafficFlowSettingsRow()
  const top = Math.min(50, Math.max(10, settings.topN))
  const windowSec = Math.max(60, q.minutes * 60)
-  const sinceIso = new Date(Date.now() - q.minutes * 60_000).toISOString()
-  const raw = listStoredFlowRows(sinceIso)
+  const raw = listFlowRowsForWindow(q.minutes)
  const allow = q.userId ? userIfaceAllow(q.userId) : null
  const serverRows = db.select().from(servers).all()
  const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host]))
+  const countryById = new Map(serverRows.map((s) => [s.id, (s.country || "").toUpperCase() || "UN"]))
+  const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : ""
+  const wantDedup = q.dedup !== false && !ifaceFilter

-  const applications = new Map<string, { bytes: number; packets: number }>()
-  const protocols = new Map<string, { bytes: number; packets: number }>()
-  const sources = new Map<string, { bytes: number; packets: number }>()
-  const destinations = new Map<string, { bytes: number; packets: number }>()
+  refreshFlowCatalogInBackground()
+
+  const applications = new Map<string, { bytes: number; packets: number; label?: string }>()
+  const protocols = new Map<string, { bytes: number; packets: number; label?: string }>()
+  const sources = new Map<string, { bytes: number; packets: number; label?: string }>()
+  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; label?: string }>()
+  const countries = new Map<string, { bytes: number; packets: number; label?: string }>()
+  const categories = new Map<string, { bytes: number; packets: number; label?: string }>()
+  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>()
  const dsts = new Set<string>()
-  let totalBytes = 0
-  let totalPackets = 0
  const matched: PendingFlowRow[] = []

  for (const r of raw) {
    const resolved = resolveIfaceName(r.serverId, r.inIface)
    if (!flowRowMatchesFilter(r, resolved.name, q, allow)) continue
    matched.push(r)
-    totalBytes += r.bytes
-    totalPackets += r.packets
-    srcs.add(r.src)
-    dsts.add(r.dst)
-    const app = applicationName(r.proto, r.dstPort, r.srcPort)
-    bump(applications, app, r.bytes, r.packets)
-    bump(protocols, protoName(r.proto), r.bytes, r.packets)
-    bump(sources, r.src, r.bytes, r.packets)
-    bump(destinations, r.dst, r.bytes, r.packets)
+
    const ifaceKey = resolved.name
    const prevIf = ifacesMap.get(ifaceKey) ?? { bytes: 0, packets: 0, index: resolved.index }
    prevIf.bytes += r.bytes
    prevIf.packets += r.packets
    ifacesMap.set(ifaceKey, prevIf)
+  }

-    const ckey = `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}|${r.inIface}`
+  const working = wantDedup ? dedupFlowRowsMaxBytes(matched) : matched
+  const conversationsRaw = new Set(matched.map((r) => `${flowTupleKey(r)}|${r.inIface}`)).size
+
+  let totalBytes = 0
+  let totalPackets = 0
+  for (const r of working) {
+    const resolved = resolveIfaceName(r.serverId, r.inIface)
+    totalBytes += r.bytes
+    totalPackets += r.packets
+    srcs.add(r.src)
+    dsts.add(r.dst)
+    const app = applicationName(r.proto, r.dstPort, r.srcPort)
+    const ripe = lookupRipeCached(r.dst)
+    const classified = classifyFlowDst(r.dst, r.proto, r.dstPort, r.srcPort, ripe)
+    bump(applications, app, r.bytes, r.packets)
+    bump(protocols, protoName(r.proto), r.bytes, r.packets)
+    bump(sources, r.src, r.bytes, r.packets)
+    bump(destinations, r.dst, r.bytes, r.packets)
+    bump(categories, classified.category, r.bytes, r.packets)
+    bump(services, classified.service, r.bytes, r.packets)
+    if (ripe?.ok && ripe.asn) {
+      const asnId = String(ripe.asn)
+      const asnLabel = ripe.holder ? `AS${ripe.asn} ${ripe.holder}` : `AS${ripe.asn}`
+      bump(asns, asnId, r.bytes, r.packets, asnLabel)
+    }
+    const dstCountry = ripe?.ok && isIsoCountry(ripe.country) ? ripe.country : ""
+    if (dstCountry) {
+      bump(countries, dstCountry, r.bytes, r.packets)
+    }
+
+    const ckey = wantDedup
+      ? flowTupleKey(r)
+      : `${flowTupleKey(r)}|${r.inIface}`
    const prev = conv.get(ckey)
    if (prev) {
      prev.rawBytes += r.bytes
@@ -152,30 +214,53 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
        inIface: resolved.name,
        inIfaceIndex: resolved.index,
        application: app,
+        category: classified.category,
+        service: classified.service,
+        dstCountry: dstCountry || undefined,
+        dstAsn: ripe?.asn || undefined,
        rawBytes: r.bytes,
      })
    }
+
+    const toCountry = dstCountry
+    if (toCountry) {
+      const fromCountry = countryById.get(r.serverId) || "UN"
+      const ekey = `${r.serverId}|${toCountry}`
+      let edge = edgeAcc.get(ekey)
+      if (!edge) {
+        edge = {
+          fromId: String(r.serverId),
+          fromLabel: nameById.get(r.serverId) ?? String(r.serverId),
+          fromCountry,
+          toCountry,
+          toAsn: ripe?.asn ?? 0,
+          category: classified.category,
+          bytes: 0,
+          bps: 0,
+          catBytes: new Map(),
+        }
+        edgeAcc.set(ekey, edge)
+      }
+      edge.bytes += r.bytes
+      if (ripe?.asn) edge.toAsn = ripe.asn
+      edge.catBytes.set(classified.category, (edge.catBytes.get(classified.category) ?? 0) + r.bytes)
+    }
  }

+  enqueueRipeMisses(dsts)
+
  const conversationsList = [...conv.values()]
    .map((t) => ({ ...t, bps: (t.rawBytes * 8) / windowSec }))
    .sort((a, b) => b.bytes - a.bytes)
    .slice(0, top)
    .map(({ rawBytes: _raw, ...rest }) => rest)

-  let topProto = "—"
-  let topProtoBytes = 0
-  for (const [label, v] of protocols) {
-    if (v.bytes > topProtoBytes) {
-      topProtoBytes = v.bytes
-      topProto = label
-    }
-  }
+  const topProto = topLabel(protocols)
+  const topCategory = topLabel(categories)

-  const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : "__all__"
  const ringServer = q.serverId ?? (matched[0]?.serverId ?? 0)
  const ring = ringServer
-    ? getRingMbps(ringServer, ifaceFilter === "__all__" ? "__all__" : (ifacesMap.get(ifaceFilter)?.index || ifaceFilter))
+    ? getRingMbps(ringServer, ifaceFilter === "" ? "__all__" : (ifacesMap.get(ifaceFilter)?.index || ifaceFilter))
    : { rx: Array(60).fill(0) as number[], tx: Array(60).fill(0) as number[], rxNow: 0, txNow: 0 }

  const fromBuckets = seriesFromRows(matched, q.minutes)
@@ -190,21 +275,46 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
      bps: (v.bytes * 8) / windowSec,
    }))

-  const protoBreakdown = topN(protocols, windowSec, top)
+  const ifaceRawBytes = [...ifacesMap.values()].reduce((a, v) => a + v.bytes, 0) || 1
  const listener = getFlowListenerState()
+  const mapEdges: FlowMapEdge[] = [...edgeAcc.values()]
+    .map((e) => {
+      let cat = e.category
+      let catBest = 0
+      for (const [label, bytes] of e.catBytes) {
+        if (bytes > catBest) {
+          catBest = bytes
+          cat = label
+        }
+      }
+      return {
+        fromId: e.fromId,
+        fromLabel: e.fromLabel,
+        fromCountry: e.fromCountry,
+        toCountry: e.toCountry,
+        toAsn: e.toAsn,
+        category: cat,
+        bytes: e.bytes,
+        bps: (e.bytes * 8) / windowSec,
+      }
+    })
+    .sort((a, b) => b.bytes - a.bytes)
+    .slice(0, top)

  return {
    bpsNow: (ring.rxNow + ring.txNow) * 1_000_000 || (totalBytes * 8) / windowSec,
    bytes: totalBytes,
    packets: totalPackets,
    conversations: conv.size,
+    conversationsRaw,
    uniqueSrc: srcs.size,
    uniqueDst: dsts.size,
    topProto,
+    topCategory,
    rxSeries,
    txSeries,
    applications: topN(applications, windowSec, top),
-    protocols: protoBreakdown,
+    protocols: topN(protocols, windowSec, top),
    sources: topN(sources, windowSec, top),
    destinations: topN(destinations, windowSec, top),
    interfaces: [...ifacesMap.entries()].map(([label, v]) => ({
@@ -213,11 +323,17 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
      bytes: v.bytes,
      packets: v.packets,
      bps: (v.bytes * 8) / windowSec,
-      percent: totalBytes > 0 ? (v.bytes / totalBytes) * 100 : 0,
+      percent: (v.bytes / ifaceRawBytes) * 100,
    })).sort((a, b) => b.bytes - a.bytes),
+    asns: topN(asns, windowSec, top),
+    countries: topN(countries, windowSec, top),
+    categories: topN(categories, windowSec, top),
+    services: topN(services, windowSec, top),
+    mapEdges,
    conversationsList,
    ifaces: ifaceRows,
    live: listener.bound,
+    dedupApplied: wantDedup,
  }
 }

@@ -245,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 = listStoredFlowRows(sinceIso)
+  const rows = listFlowRowsForWindow(minutes)
  const ids = new Set<number>()
  for (const r of rows) ids.add(r.serverId)
  for (const p of listHostPeers()) ids.add(p.serverId)
@@ -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")
@@ -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)
+}
@@ -0,0 +1,25 @@
+import assert from "node:assert/strict"
+import { classifyFlowDst, disableCatalogFetchForTests, resetFlowCatalogForTests, seedFlowCatalogForTests } from "./traffic-flow-classify.js"
+
+disableCatalogFetchForTests()
+resetFlowCatalogForTests()
+seedFlowCatalogForTests({
+  cidrs: [{ cidr: "192.0.2.0/24", purpose: "steam-gaming" }],
+})
+
+const hit = classifyFlowDst("192.0.2.10", 6, 443, 50000, null)
+assert.equal(hit.category, "Игры")
+assert.equal(hit.service, "steam-gaming")
+
+const miss = classifyFlowDst("203.0.113.9", 17, 53, 53000, null)
+assert.equal(miss.category, "DNS")
+
+const cdn = classifyFlowDst("203.0.113.9", 6, 443, 1, { prefix: "203.0.113.0/24", asn: 13335, country: "US", lat: null, lng: null, holder: "CLOUDFLARENET", ok: true, fetchedAt: Date.now() })
+assert.equal(cdn.category, "CDN")
+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")
@@ -0,0 +1,142 @@
+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"
+import type { FlowIpMeta } from "./traffic-flow-ripe.js"
+import { applicationName } from "./traffic-flow-apps.js"
+
+export interface FlowClassification {
+  service: string
+  category: string
+}
+
+interface CatalogCidr {
+  cidr: string
+  purpose: string
+  prefixLen: number
+}
+
+const CATALOG_TTL_MS = 10 * 60_000
+let cidrs: CatalogCidr[] = []
+let asnPurpose = new Map<number, string>()
+let fetchedAt = 0
+let catalogFetchEnabled = true
+let inflight: Promise<void> | null = null
+
+export function disableCatalogFetchForTests(): void {
+  catalogFetchEnabled = false
+}
+
+export function resetFlowCatalogForTests(): void {
+  cidrs = []
+  asnPurpose = new Map()
+  fetchedAt = 0
+  inflight = null
+}
+
+export function seedFlowCatalogForTests(input: {
+  cidrs?: Array<{ cidr: string; purpose: string }>
+  asns?: Array<{ asn: number; purpose: string }>
+}): void {
+  cidrs = (input.cidrs ?? [])
+    .map((c) => ({ cidr: c.cidr, purpose: c.purpose, prefixLen: parseCidrV4(c.cidr)?.prefixLen ?? 0 }))
+    .sort((a, b) => b.prefixLen - a.prefixLen)
+  asnPurpose = new Map((input.asns ?? []).map((a) => [a.asn, a.purpose]))
+  fetchedAt = Date.now()
+}
+
+export function categoryFromPurpose(purpose: string, proto: number, dstPort: number, srcPort: number): string {
+  const p = purpose.toLowerCase()
+  if (/gaming|steam|epic|riot/.test(p)) return "Игры"
+  if (/streaming|youtube|netflix|twitch|video/.test(p)) return "Видео / стриминг"
+  if (/cdn|cloudflare|akamai|fastly/.test(p)) return "CDN"
+  if (/voip|discord|zoom/.test(p)) return "Голос"
+  if (/openai|chatgpt|\bai\b/.test(p)) return "ИИ"
+  const app = applicationName(proto, dstPort, srcPort)
+  if (app === "DNS" || app === "SSH" || app === "BGP") return app
+  return OTHER_SERVICE
+}
+
+function matchCidr(ip: string): CatalogCidr | null {
+  for (const row of cidrs) {
+    if (ipInCidrV4(ip, row.cidr)) return row
+  }
+  return null
+}
+
+export function classifyFlowDst(
+  dst: string,
+  proto: number,
+  dstPort: number,
+  srcPort: number,
+  ripe: FlowIpMeta | null,
+): FlowClassification {
+  const hit = matchCidr(dst)
+  const brand = lookupBrand(dst, ripe?.asn ?? 0)
+  const asnName = ripe?.asn ? asnPurpose.get(ripe.asn) : undefined
+  const service = (hit?.purpose || brand?.service || asnName || OTHER_SERVICE).trim() || OTHER_SERVICE
+  const category = hit
+    ? categoryFromPurpose(hit.purpose, proto, dstPort, srcPort)
+    : (brand?.category || categoryFromPurpose(asnName || "", proto, dstPort, srcPort))
+  return { service, category }
+}
+
+async function fetchCatalog(): Promise<void> {
+  if (!catalogFetchEnabled) return
+  if (Date.now() - fetchedAt < CATALOG_TTL_MS) return
+  if (inflight) return inflight
+  inflight = (async () => {
+    try {
+      const row = db.select().from(evobgpSettings).limit(1).all()[0]
+      if (!row?.enabled) return
+      const root = String(row.baseUrl ?? "").replace(/\/+$/, "")
+      const token = String(row.apiKey ?? "").replace(/^Bearer\s+/i, "").trim()
+      if (!root || !token) return
+      const ac = new AbortController()
+      const t = setTimeout(() => ac.abort(), 20_000)
+      try {
+        const res = await fetch(`${root}/v1/router-lists/catalog`, {
+          headers: { Authorization: `Bearer ${token}`, Accept: "application/json" },
+          signal: ac.signal,
+        })
+        if (!res.ok) return
+        const catalog = await res.json() as {
+          modules?: { items?: Array<{ id: string; name: string }> }
+          ip_ranges?: { items?: Array<{ module_id: string; entry: { prefix: string } }> }
+          asns?: { items?: Array<{ module_id: string; entry: { asn: number } }> }
+        }
+        const mods = new Map((catalog.modules?.items ?? []).map((m) => [m.id, m.name]))
+        const next: CatalogCidr[] = []
+        for (const item of catalog.ip_ranges?.items ?? []) {
+          const prefix = String(item.entry?.prefix ?? "").trim()
+          const purpose = mods.get(item.module_id) ?? ""
+          const parsed = parseCidrV4(prefix)
+          if (!prefix || !parsed) continue
+          next.push({ cidr: prefix, purpose, prefixLen: parsed.prefixLen })
+        }
+        next.sort((a, b) => b.prefixLen - a.prefixLen)
+        const nextAsn = new Map<number, string>()
+        for (const item of catalog.asns?.items ?? []) {
+          const purpose = mods.get(item.module_id)
+          const asn = Number(item.entry?.asn)
+          if (purpose && Number.isFinite(asn) && asn > 0) nextAsn.set(asn, purpose)
+        }
+        cidrs = next
+        asnPurpose = nextAsn
+        fetchedAt = Date.now()
+      } finally {
+        clearTimeout(t)
+      }
+    } catch {
+      /* catalog optional */
+    } finally {
+      inflight = null
+    }
+  })()
+  return inflight
+}
+
+/** Background refresh — analytics never awaits the HTTP. */
+export function refreshFlowCatalogInBackground(): void {
+  void fetchCatalog()
+}
@@ -0,0 +1,25 @@
+import assert from "node:assert/strict"
+import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js"
+
+const a = {
+  serverId: 7,
+  src: "10.1.1.8",
+  dst: "8.8.8.8",
+  proto: 6,
+  srcPort: 1,
+  dstPort: 443,
+  inIface: "2",
+  bytes: 12_000,
+  packets: 10,
+}
+const b = { ...a, inIface: "10", bytes: 8_000, packets: 8 }
+const out = dedupFlowRowsMaxBytes([a, b])
+assert.equal(out.length, 1)
+assert.equal(out[0]?.bytes, 12_000)
+assert.equal(out[0]?.inIface, "2")
+assert.equal(flowTupleKey(a), flowTupleKey(b))
+
+const sameIface = dedupFlowRowsMaxBytes([a, { ...a, bytes: 3_000, packets: 2 }])
+assert.equal(sameIface[0]?.bytes, 15_000)
+
+console.log("traffic-flow-dedup.test.ts: ok")
@@ -0,0 +1,44 @@
+export interface FlowTupleRow {
+  serverId: number
+  src: string
+  dst: string
+  proto: number
+  srcPort: number
+  dstPort: number
+  inIface: string
+  bytes: number
+  packets: number
+}
+
+export function flowTupleKey(r: Pick<FlowTupleRow, "serverId" | "src" | "dst" | "proto" | "srcPort" | "dstPort">): string {
+  return `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}`
+}
+
+function ifaceKey(r: FlowTupleRow): string {
+  return `${flowTupleKey(r)}|${r.inIface}`
+}
+
+/**
+ * Один 5-tuple на двух ifIndex — это один поток: сначала сумма по бакетам/iface,
+ * затем max байт между интерфейсами (не sum).
+ */
+export function dedupFlowRowsMaxBytes<T extends FlowTupleRow>(rows: T[]): T[] {
+  const byIface = new Map<string, T>()
+  for (const row of rows) {
+    const key = ifaceKey(row)
+    const prev = byIface.get(key)
+    if (!prev) {
+      byIface.set(key, { ...row })
+      continue
+    }
+    prev.bytes += row.bytes
+    prev.packets += row.packets
+  }
+  const byTuple = new Map<string, T>()
+  for (const row of byIface.values()) {
+    const key = flowTupleKey(row)
+    const prev = byTuple.get(key)
+    if (!prev || row.bytes > prev.bytes) byTuple.set(key, row)
+  }
+  return [...byTuple.values()]
+}
@@ -4,6 +4,8 @@ import {
  resetIfaceCacheForTests,
  resolveIfaceName,
  rosIdToIfIndex,
+  shouldRefreshIfaces,
+  markIfaceRefreshAttempt,
 } from "./traffic-flow-ifindex.js"
 import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"

@@ -27,6 +29,22 @@ assert.equal(resolveIfaceName(7, "0").name, "—")
 assert.equal(resolveIfaceName(7, "ether1").name, "ether1")
 assert.equal(resolveIfaceName(7, "99").name, "#99")

+resetIfaceCacheForTests()
+rememberServerIfaces(8, [
+  { ifindex: "10", ".id": "*12", name: "gre1" },
+])
+assert.equal(rosIdToIfIndex("*12"), 18)
+assert.equal(resolveIfaceName(8, "10").name, "gre1")
+assert.equal(resolveIfaceName(8, "18").name, "gre1")
+
+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")
@@ -3,24 +3,28 @@ 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"

 export {
+  ifaceCacheFresh,
  ifaceCacheHas,
  rememberServerIfaces,
  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]
@@ -31,6 +35,7 @@ export async function refreshServerIfaces(serverId: number, force = false): Prom
  } catch {
    /* keep previous cache */
  } finally {
+    markIfaceRefreshAttempt(serverId)
    inflight.delete(serverId)
  }
 }
@@ -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

@@ -25,10 +26,9 @@ export function rememberServerIfaces(serverId: number, rows: RosIfaceIndexRow[])
    const name = String(row.name ?? "").trim()
    if (!name) continue
    const fromProp = Number.parseInt(String(row.ifindex ?? ""), 10)
-    const idx = Number.isFinite(fromProp) && fromProp > 0
-      ? fromProp
-      : rosIdToIfIndex(row[".id"])
-    if (idx != null && idx > 0) map.set(idx, name)
+    const fromId = rosIdToIfIndex(row[".id"])
+    if (Number.isFinite(fromProp) && fromProp > 0) map.set(fromProp, name)
+    if (fromId != null && fromId > 0) map.set(fromId, name)
  }
  cache.set(serverId, map)
  fetchedAt.set(serverId, Date.now())
@@ -53,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()
 }
@@ -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")
@@ -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 { ifaceCacheHas, 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,10 +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
-  if (!ifaceCacheHas(serverId)) void refreshServerIfaces(serverId)
+  maybeRefreshIfaces(serverId)
  const bucketAt = minuteBucketIso()
  for (const flow of flows) {
    addToTick(serverId, flow.inIface, flow.outIface, flow.bytes)
@@ -168,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",
@@ -180,53 +268,17 @@ export function peekPendingFlows(): PendingFlowRow[] {
    packets: row.packets,
    inIface: row.flow.inIface,
    outIface: row.flow.outIface,
-  }))
+  }
 }

-function flushPending() {
-  if (pending.size === 0) return
+function pruneStoredBuckets(): void {
+  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 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
-    }
-  }
-
+  const cutoff = new Date(now - settings.retentionHours * 3600_000).toISOString()
  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
@@ -246,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()
@@ -304,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}`
-    merged.set(key, {
+    mergeInto(merged, {
      serverId: r.serverId,
      bucketAt: r.bucketAt,
      src: r.src,
@@ -325,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}`
-    const prev = merged.get(key)
-    if (prev) {
-      prev.bytes += p.bytes
-      prev.packets += p.packets
-    } else {
-      merged.set(key, { ...p })
-    }
+    mergeInto(merged, 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 = listStoredFlowRows(rangeStart)
+  const rows = listFlowRowsForWindow(minutes)
  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 }>()
@@ -424,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
@@ -446,4 +561,8 @@ export function resetFlowRingsForTests() {
  tickAccum.clear()
  rings.clear()
  pending.clear()
+  recent.clear()
+  lastPruneAt = 0
+  lastFlushUsedTransaction = false
+  refreshIfacesImpl = refreshServerIfaces
 }
@@ -0,0 +1,54 @@
+/** IPv4 helpers for RIPEstat prefix cache and EvoBGP CIDR match. */
+
+export function ipv4ToInt(ip: string): number | null {
+  const parts = String(ip ?? "").trim().split(".")
+  if (parts.length !== 4) return null
+  let n = 0
+  for (const p of parts) {
+    if (!/^\d+$/.test(p)) return null
+    const o = Number(p)
+    if (o < 0 || o > 255) return null
+    n = ((n << 8) >>> 0) + o
+  }
+  return n >>> 0
+}
+
+export function parseCidrV4(cidr: string): { net: number; mask: number; prefixLen: number } | null {
+  const raw = String(cidr ?? "").trim()
+  const [ip, lenRaw] = raw.split("/")
+  const addr = ipv4ToInt(ip ?? "")
+  const prefixLen = Number.parseInt(lenRaw ?? "", 10)
+  if (addr == null || !Number.isFinite(prefixLen) || prefixLen < 0 || prefixLen > 32) return null
+  const mask = prefixLen === 0 ? 0 : (0xffffffff << (32 - prefixLen)) >>> 0
+  return { net: (addr & mask) >>> 0, mask, prefixLen }
+}
+
+export function ipInCidrV4(ip: string, cidr: string): boolean {
+  const addr = ipv4ToInt(ip)
+  const parsed = parseCidrV4(cidr)
+  if (addr == null || !parsed) return false
+  return ((addr & parsed.mask) >>> 0) === parsed.net
+}
+
+export function isNonPublicIp(ip: string): boolean {
+  const trimmed = String(ip ?? "").trim()
+  if (!trimmed) return true
+  if (trimmed.includes(":")) {
+    const lower = trimmed.toLowerCase()
+    return lower === "::1" || lower.startsWith("fe80:") || lower.startsWith("fc") || lower.startsWith("fd") || lower === "::"
+  }
+  const n = ipv4ToInt(trimmed)
+  if (n == null) return true
+  const inRange = (cidr: string) => ipInCidrV4(trimmed, cidr)
+  return (
+    inRange("0.0.0.0/8")
+    || inRange("10.0.0.0/8")
+    || inRange("127.0.0.0/8")
+    || inRange("169.254.0.0/16")
+    || inRange("172.16.0.0/12")
+    || inRange("192.168.0.0/16")
+    || inRange("100.64.0.0/10")
+    || inRange("224.0.0.0/4")
+    || inRange("255.255.255.255/32")
+  )
+}
@@ -96,7 +96,7 @@ resetFlowTemplatesForTests()
  parseFlowPacket(tpl, "10.255.254.3")
  const named = parseFlowPacket(data, "10.255.254.3")
  assert.equal(named.length, 1)
-  assert.equal(named[0]?.inIface, "13")
+  assert.equal(named[0]?.inIface, "ether1")
  assert.equal(named[0]?.src, "10.1.1.8")
 }

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

@@ -0,0 +1,103 @@
+import assert from "node:assert/strict"
+import {
+  disableRipeEnqueueForTests,
+  disableRipePersistForTests,
+  enqueueRipeMisses,
+  flushRipeQueueForTests,
+  lookupRipeCached,
+  resetRipeCacheForTests,
+  ripeFetchCountForTests,
+  seedRipeCacheForTests,
+  setRipeFetchForTests,
+} from "./traffic-flow-ripe.js"
+
+disableRipePersistForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+
+assert.equal(lookupRipeCached("10.1.1.8")?.ok, false)
+assert.equal(lookupRipeCached("192.168.0.1")?.ok, false)
+assert.equal(lookupRipeCached("100.64.1.2")?.ok, false)
+assert.equal(ripeFetchCountForTests(), 0)
+
+seedRipeCacheForTests({
+  prefix: "1.2.3.0/24",
+  asn: 64500,
+  country: "NL",
+  lat: 52.3,
+  lng: 4.9,
+  holder: "TEST",
+  ok: true,
+  fetchedAt: Date.now(),
+})
+assert.equal(lookupRipeCached("1.2.3.10")?.country, "NL")
+assert.equal(lookupRipeCached("1.2.3.10")?.asn, 64500)
+assert.equal(ripeFetchCountForTests(), 0)
+
+resetRipeCacheForTests()
+disableRipePersistForTests()
+setRipeFetchForTests(async (input) => {
+  const url = String(input)
+  const body = url.includes("network-info")
+    ? { data: { prefix: "8.8.8.0/24", asns: ["15169"] } }
+    : url.includes("maxmind-geo-lite")
+      ? { data: { located_resources: [{ locations: [{ country: "US", latitude: 37.4, longitude: -122.1 }] }] } }
+      : { data: { holder: "GOOGLE" } }
+  return new Response(JSON.stringify(body), { status: 200, headers: { "Content-Type": "application/json" } })
+})
+enqueueRipeMisses(["8.8.8.8"])
+await flushRipeQueueForTests()
+assert.equal(lookupRipeCached("8.8.8.8")?.country, "US")
+assert.equal(lookupRipeCached("8.8.8.10")?.prefix, "8.8.8.0/24")
+const afterFirst = ripeFetchCountForTests()
+assert.ok(afterFirst >= 2)
+enqueueRipeMisses(["8.8.8.10"])
+await flushRipeQueueForTests()
+assert.equal(ripeFetchCountForTests(), afterFirst)
+
+resetRipeCacheForTests()
+disableRipePersistForTests()
+setRipeFetchForTests(async () => {
+  throw new Error("timeout")
+})
+enqueueRipeMisses(["203.0.113.50"])
+await flushRipeQueueForTests()
+const neg = lookupRipeCached("203.0.113.50")
+assert.equal(neg?.ok, false)
+const afterNeg = ripeFetchCountForTests()
+enqueueRipeMisses(["203.0.113.50"])
+await flushRipeQueueForTests()
+assert.equal(ripeFetchCountForTests(), afterNeg)
+
+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")
@@ -0,0 +1,379 @@
+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
+  asn: number
+  country: string
+  lat: number | null
+  lng: number | null
+  holder: string
+  ok: boolean
+  fetchedAt: number
+}
+
+const HIT_TTL_MS = 24 * 60 * 60_000
+const NEG_TTL_MS = 6 * 60 * 60_000
+const MAX_NEW_PREFIX_PER_MIN = 30
+const MAX_QUEUE = 90
+const CONCURRENCY = 3
+const RIPE_BASE = "https://stat.ripe.net/data"
+const UA = "MikrotikManager-flow/1.0"
+
+const mem = new Map<string, FlowIpMeta>()
+const asnHolder = new Map<number, { holder: string; fetchedAt: number }>()
+const inflight = new Map<string, Promise<FlowIpMeta | null>>()
+const queue: string[] = []
+const queued = new Set<string>()
+const recentFetches: number[] = []
+
+let persistEnabled = true
+let enqueueEnabled = true
+let loaded = false
+let workerRunning = false
+let fetchImpl: typeof fetch = globalThis.fetch.bind(globalThis)
+let fetchCount = 0
+
+export function disableRipePersistForTests(): void {
+  persistEnabled = false
+}
+
+export function disableRipeEnqueueForTests(): void {
+  enqueueEnabled = false
+}
+
+export function resetRipeCacheForTests(): void {
+  mem.clear()
+  asnHolder.clear()
+  inflight.clear()
+  queue.length = 0
+  queued.clear()
+  recentFetches.length = 0
+  loaded = persistEnabled ? false : true
+  workerRunning = false
+  fetchCount = 0
+  enqueueEnabled = true
+  fetchImpl = globalThis.fetch.bind(globalThis)
+}
+
+export function seedRipeCacheForTests(entry: FlowIpMeta): void {
+  mem.set(entry.prefix, { ...entry })
+  loaded = true
+}
+
+export function setRipeFetchForTests(fn: typeof fetch): void {
+  fetchImpl = fn
+  fetchCount = 0
+}
+
+export function ripeFetchCountForTests(): number {
+  return fetchCount
+}
+
+export async function flushRipeQueueForTests(timeoutMs = 4000): Promise<void> {
+  const start = Date.now()
+  while (Date.now() - start < timeoutMs) {
+    if (!queue.length && !inflight.size && !workerRunning) return
+    await new Promise((r) => setTimeout(r, 20))
+  }
+}
+
+function ttlMs(ok: boolean): number {
+  return ok ? HIT_TTL_MS : NEG_TTL_MS
+}
+
+function isFresh(entry: FlowIpMeta): boolean {
+  return Date.now() - entry.fetchedAt < ttlMs(entry.ok)
+}
+
+function loadSqlite(): void {
+  if (loaded || !persistEnabled) {
+    loaded = true
+    return
+  }
+  loaded = true
+  try {
+    const rows = sqliteDatabase.prepare(`
+      SELECT prefix, asn, country, lat, lng, holder, ok, fetched_at
+      FROM flow_ip_meta
+    `).all() as Array<{
+      prefix: string
+      asn: number | null
+      country: string
+      lat: number | null
+      lng: number | null
+      holder: string
+      ok: number
+      fetched_at: string
+    }>
+    for (const r of rows) {
+      const fetchedAt = Date.parse(r.fetched_at)
+      const asn = Number(r.asn ?? 0) || 0
+      const holder = r.holder || ""
+      mem.set(r.prefix, {
+        prefix: r.prefix,
+        asn,
+        country: resolveRipeCountry(r.country || "", asn, holder) || "—",
+        lat: r.lat == null ? null : Number(r.lat),
+        lng: r.lng == null ? null : Number(r.lng),
+        holder,
+        ok: r.ok !== 0,
+        fetchedAt: Number.isFinite(fetchedAt) ? fetchedAt : 0,
+      })
+    }
+    const asns = sqliteDatabase.prepare(`SELECT asn, holder, fetched_at FROM flow_asn_meta`).all() as Array<{
+      asn: number
+      holder: string
+      fetched_at: string
+    }>
+    for (const a of asns) {
+      const fetchedAt = Date.parse(a.fetched_at)
+      asnHolder.set(a.asn, { holder: a.holder || "", fetchedAt: Number.isFinite(fetchedAt) ? fetchedAt : 0 })
+    }
+  } catch {
+    /* table may not exist in isolated tests */
+  }
+}
+
+function persist(entry: FlowIpMeta): void {
+  if (!persistEnabled) return
+  try {
+    sqliteDatabase.prepare(`
+      INSERT INTO flow_ip_meta (prefix, asn, country, lat, lng, holder, ok, fetched_at)
+      VALUES (@prefix, @asn, @country, @lat, @lng, @holder, @ok, @fetchedAt)
+      ON CONFLICT(prefix) DO UPDATE SET
+        asn=excluded.asn, country=excluded.country, lat=excluded.lat, lng=excluded.lng,
+        holder=excluded.holder, ok=excluded.ok, fetched_at=excluded.fetched_at
+    `).run({
+      prefix: entry.prefix,
+      asn: entry.asn,
+      country: entry.country,
+      lat: entry.lat,
+      lng: entry.lng,
+      holder: entry.holder,
+      ok: entry.ok ? 1 : 0,
+      fetchedAt: new Date(entry.fetchedAt).toISOString(),
+    })
+  } catch {
+    /* ignore persist errors */
+  }
+}
+
+function persistAsn(asn: number, holder: string): void {
+  if (!persistEnabled || !asn) return
+  try {
+    sqliteDatabase.prepare(`
+      INSERT INTO flow_asn_meta (asn, holder, fetched_at)
+      VALUES (@asn, @holder, @fetchedAt)
+      ON CONFLICT(asn) DO UPDATE SET holder=excluded.holder, fetched_at=excluded.fetched_at
+    `).run({
+      asn,
+      holder,
+      fetchedAt: new Date().toISOString(),
+    })
+  } catch {
+    /* ignore */
+  }
+}
+
+function negative(prefix: string): FlowIpMeta {
+  return {
+    prefix,
+    asn: 0,
+    country: "—",
+    lat: null,
+    lng: null,
+    holder: "",
+    ok: false,
+    fetchedAt: Date.now(),
+  }
+}
+
+export function lookupRipeCached(ip: string): FlowIpMeta | null {
+  loadSqlite()
+  const trimmed = String(ip ?? "").trim()
+  if (!trimmed) return null
+  if (isNonPublicIp(trimmed)) {
+    return negative(`${trimmed.includes(":") ? trimmed : trimmed}/32`)
+  }
+  let best: FlowIpMeta | null = null
+  let bestLen = -1
+  for (const entry of mem.values()) {
+    if (!isFresh(entry)) continue
+    const parsed = parseCidrV4(entry.prefix)
+    if (!parsed) continue
+    if (!ipInCidrV4(trimmed, entry.prefix)) continue
+    if (parsed.prefixLen > bestLen) {
+      best = entry
+      bestLen = parsed.prefixLen
+    }
+  }
+  return best
+    ? { ...best, country: resolveRipeCountry(best.country, best.asn, best.holder) || "—" }
+    : null
+}
+
+async function ripeJson(path: string, resource: string): Promise<unknown> {
+  fetchCount += 1
+  const url = `${RIPE_BASE}/${path}/data.json?resource=${encodeURIComponent(resource)}`
+  const ac = new AbortController()
+  const t = setTimeout(() => ac.abort(), 12_000)
+  try {
+    const res = await fetchImpl(url, {
+      headers: { Accept: "application/json", "User-Agent": UA },
+      signal: ac.signal,
+    })
+    if (!res.ok) throw new Error(`HTTP ${res.status}`)
+    return await res.json()
+  } finally {
+    clearTimeout(t)
+  }
+}
+
+function pickPrefix(data: unknown): string {
+  const d = data as { data?: { prefix?: string } }
+  return String(d?.data?.prefix ?? "").trim()
+}
+
+function pickAsns(data: unknown): number {
+  const d = data as { data?: { asns?: unknown } }
+  const raw = d?.data?.asns
+  const first = Array.isArray(raw) ? raw[0] : raw
+  const n = Number.parseInt(String(first ?? "").replace(/^AS/i, ""), 10)
+  return Number.isFinite(n) ? n : 0
+}
+
+function pickGeo(data: unknown): { country: string; lat: number | null; lng: number | null } {
+  const d = data as {
+    data?: {
+      located_resources?: Array<{
+        locations?: Array<{ country?: string; latitude?: number; longitude?: number }>
+      }>
+    }
+  }
+  const loc = d?.data?.located_resources?.[0]?.locations?.[0]
+  const country = resolveRipeCountry(String(loc?.country ?? ""), 0, "")
+  const lat = loc?.latitude == null ? null : Number(loc.latitude)
+  const lng = loc?.longitude == null ? null : Number(loc.longitude)
+  return {
+    country: country || "—",
+    lat: Number.isFinite(lat) ? lat : null,
+    lng: Number.isFinite(lng) ? lng : null,
+  }
+}
+
+function pickHolder(data: unknown): string {
+  const d = data as { data?: { holder?: string } }
+  return String(d?.data?.holder ?? "").trim()
+}
+
+function allowNewPrefix(): boolean {
+  const now = Date.now()
+  while (recentFetches.length && now - recentFetches[0]! > 60_000) recentFetches.shift()
+  return recentFetches.length < MAX_NEW_PREFIX_PER_MIN
+}
+
+async function resolveIp(ip: string): Promise<FlowIpMeta | null> {
+  const cached = lookupRipeCached(ip)
+  if (cached) return cached
+  const pending = inflight.get(ip)
+  if (pending) return pending
+
+  const job = (async () => {
+    if (!allowNewPrefix()) return null
+    recentFetches.push(Date.now())
+    try {
+      const net = await ripeJson("network-info", ip)
+      const prefix = pickPrefix(net) || `${ip}/32`
+      const existing = mem.get(prefix)
+      if (existing && isFresh(existing)) return existing
+      const asn = pickAsns(net)
+      let geo = { country: "—", lat: null as number | null, lng: null as number | null }
+      try {
+        geo = pickGeo(await ripeJson("maxmind-geo-lite", prefix))
+      } catch {
+        /* best-effort */
+      }
+      let holder = asnHolder.get(asn)?.holder ?? ""
+      if (asn && (!holder || Date.now() - (asnHolder.get(asn)?.fetchedAt ?? 0) > HIT_TTL_MS)) {
+        try {
+          holder = pickHolder(await ripeJson("as-overview", `AS${asn}`))
+          asnHolder.set(asn, { holder, fetchedAt: Date.now() })
+          persistAsn(asn, holder)
+        } catch {
+          /* best-effort */
+        }
+      }
+      const country = resolveRipeCountry(geo.country, asn, holder)
+      const entry: FlowIpMeta = {
+        prefix,
+        asn,
+        country: country || "—",
+        lat: geo.lat,
+        lng: geo.lng,
+        holder,
+        ok: Boolean(asn || country),
+        fetchedAt: Date.now(),
+      }
+      mem.set(prefix, entry)
+      persist(entry)
+      return entry
+    } catch {
+      const prefix = `${ip}/32`
+      const entry = negative(prefix)
+      mem.set(prefix, entry)
+      persist(entry)
+      return entry
+    } finally {
+      inflight.delete(ip)
+    }
+  })()
+
+  inflight.set(ip, job)
+  return job
+}
+
+async function runWorker(): Promise<void> {
+  if (workerRunning) return
+  workerRunning = true
+  try {
+    while (queue.length) {
+      const batch: string[] = []
+      while (batch.length < CONCURRENCY && queue.length) {
+        const ip = queue.shift()
+        if (!ip) break
+        queued.delete(ip)
+        if (lookupRipeCached(ip)) continue
+        if (ipv4ToInt(ip) == null && !ip.includes(":")) continue
+        batch.push(ip)
+      }
+      if (!batch.length) {
+        if (!allowNewPrefix()) {
+          await new Promise((r) => setTimeout(r, 1000))
+        }
+        continue
+      }
+      await Promise.all(batch.map((ip) => resolveIp(ip)))
+    }
+  } finally {
+    workerRunning = false
+    if (queue.length) void runWorker()
+  }
+}
+
+/** HTTP / SSE never await this — cache miss is filled on a later tick. */
+export function enqueueRipeMisses(ips: Iterable<string>): void {
+  if (!enqueueEnabled) return
+  loadSqlite()
+  for (const raw of ips) {
+    if (queue.length >= MAX_QUEUE) break
+    const ip = String(raw ?? "").trim()
+    if (!ip || isNonPublicIp(ip)) continue
+    if (lookupRipeCached(ip)) continue
+    if (queued.has(ip) || inflight.has(ip)) continue
+    queued.add(ip)
+    queue.push(ip)
+  }
+  if (queue.length) void runWorker()
+}
@@ -64,7 +64,14 @@ function TrafficFlowsDataGrid({
        accessorFn: (r) => r.application ?? r.protoName,
        header: () => <span className="text-xs font-medium text-muted-foreground">App</span>,
        cell: ({ row }) => (
-          <span className="text-xs">{row.original.application ?? row.original.protoName}</span>
+          <span className="flex min-w-0 flex-col gap-0.5 text-xs">
+            <span>{row.original.application ?? row.original.protoName}</span>
+            {row.original.category || row.original.service ? (
+              <span className="text-[10px] text-muted-foreground truncate">
+                {[row.original.category, row.original.service].filter(Boolean).join(" · ")}
+              </span>
+            ) : null}
+          </span>
        ),
        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
      },
@@ -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)
@@ -2,12 +2,14 @@

 import { useMemo, useState } from "react"
 import { type ColumnDef, getCoreRowModel, useReactTable } from "@tanstack/react-table"
-import type { FlowAnalyticsDto, FlowBreakdownRow, FlowEntityCard } from "@mmapp/contracts/traffic-flow"
-import { ArrowDownIcon, ArrowUpIcon, GitBranchIcon } from "lucide-react"
+import 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"
 import { TrafficRxTxChart } from "@/components/reui-kit/traffic-rx-tx-chart"
 import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"
+import { Switch } from "@/components/ui/switch"
+import { Label } from "@/components/ui/label"
 import { DataGridShell } from "@/components/data-grids/shared/data-grid-shell"
 import {
  DATA_GRID_CELL_PAD,
@@ -15,7 +17,7 @@ import {
  DATA_GRID_CELL_PAD_LAST,
 } from "@/components/data-grids/shared/data-grid-layout"
 import { TrafficFlowsDataGrid } from "@/components/data-grids/traffic-flows-data-grid"
-import { Sparkline } from "@/components/sparkline"
+import { FlowTrafficMap } from "@/components/traffic/flow-traffic-map"
 import { StatusDot } from "@/components/status-dot"
 import { Flag } from "@/components/flag"
 import { fmtRate } from "@/lib/fmt-rate"
@@ -106,7 +108,37 @@ export function FlowEntityCardView({
  )
 }

-function FlowBreakdownGrid({ rows, empty }: { rows: FlowBreakdownRow[]; empty?: string }) {
+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>[]>(
    () => [
      {
@@ -115,7 +147,10 @@ function FlowBreakdownGrid({ rows, empty }: { rows: FlowBreakdownRow[]; empty?:
        header: () => <span className="text-xs font-medium text-muted-foreground">Имя</span>,
        cell: ({ row }) => (
          <div className="flex min-w-0 flex-col gap-1">
-            <span className="text-sm font-medium truncate">{row.original.label}</span>
+            <span className="flex items-center gap-1.5 text-sm font-medium truncate">
+              {country ? <Flag code={row.original.id} /> : null}
+              {row.original.label}
+            </span>
            <div className="h-1 rounded-full bg-muted overflow-hidden">
              <div
                className="h-full bg-primary"
@@ -154,7 +189,7 @@ function FlowBreakdownGrid({ rows, empty }: { rows: FlowBreakdownRow[]; empty?:
        meta: { headerClassName: DATA_GRID_CELL_PAD_LAST, cellClassName: DATA_GRID_CELL_PAD_LAST },
      },
    ],
-    [],
+    [country],
  )

  const table = useReactTable({
@@ -165,7 +200,12 @@ function FlowBreakdownGrid({ rows, empty }: { rows: FlowBreakdownRow[]; empty?:
  })

  return (
-    <DataGridShell table={table} recordCount={rows.length} emptyMessage={empty ?? "Нет данных за период"} />
+    <DataGridShell
+      table={table}
+      recordCount={rows.length}
+      emptyMessage={empty ?? "Нет данных за период"}
+      onRowClick={onPick}
+    />
  )
 }

@@ -176,6 +216,8 @@ export function FlowAnalyticsDetail({
  onRange,
  selectedIface,
  onIface,
+  dedup,
+  onDedup,
  liveHint,
  emptyHint,
 }: {
@@ -185,12 +227,23 @@ export function FlowAnalyticsDetail({
  onRange: (r: string) => void
  selectedIface: string
  onIface: (name: string) => void
+  dedup: boolean
+  onDedup: (value: boolean) => void
  liveHint?: string
  emptyHint?: string
 }) {
  const [slice, setSlice] = useState("applications")
+  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) =>
+    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 (
@@ -211,22 +264,34 @@ export function FlowAnalyticsDetail({
            {card.site}
          </span>
        </div>
-        <div className="flex gap-1 shrink-0">
-          {RANGE_KEYS.map((r) => (
-            <button
-              key={r}
-              type="button"
-              onClick={() => onRange(r)}
-              className={cn(
-                "h-7 px-2 text-xs rounded border transition-colors",
-                range === r
-                  ? "border-primary bg-primary/10 text-primary font-medium"
-                  : "border-border text-muted-foreground hover:text-foreground",
-              )}
-            >
-              {RANGE_LABELS[r]}
-            </button>
-          ))}
+        <div className="flex items-center gap-3 shrink-0">
+          <div className="flex items-center gap-2">
+            <Switch
+              id="flow-dedup"
+              checked={dedup}
+              onCheckedChange={onDedup}
+            />
+            <Label htmlFor="flow-dedup" className="text-xs text-muted-foreground whitespace-nowrap">
+              Без дублей
+            </Label>
+          </div>
+          <div className="flex gap-1">
+            {RANGE_KEYS.map((r) => (
+              <button
+                key={r}
+                type="button"
+                onClick={() => onRange(r)}
+                className={cn(
+                  "h-7 px-2 text-xs rounded border transition-colors",
+                  range === r
+                    ? "border-primary bg-primary/10 text-primary font-medium"
+                    : "border-border text-muted-foreground hover:text-foreground",
+                )}
+              >
+                {RANGE_LABELS[r]}
+              </button>
+            ))}
+          </div>
        </div>
      </div>

@@ -265,6 +330,7 @@ export function FlowAnalyticsDetail({
              )
            })}
          </div>
+          <p className="text-[10px] text-muted-foreground mt-1.5">по iface, без дедупа</p>
        </div>
      ) : null}

@@ -295,11 +361,28 @@ export function FlowAnalyticsDetail({
            },
            {
              id: "flows",
-              label: "Разговоры",
+              label: "Сессии",
              value: String(analytics?.conversations ?? card.sessions),
+              hint: analytics?.conversationsRaw != null && analytics.conversationsRaw !== analytics.conversations
+                ? `до дедупа ${analytics.conversationsRaw}`
+                : undefined,
              icon: <GitBranchIcon className="size-4" />,
              iconClassName: "text-warning",
            },
+            {
+              id: "uniq",
+              label: "Уник. src / dst",
+              value: `${analytics?.uniqueSrc ?? 0} / ${analytics?.uniqueDst ?? 0}`,
+              icon: <UsersIcon className="size-4" />,
+              iconClassName: "text-muted-foreground",
+            },
+            {
+              id: "category",
+              label: "Топ категория",
+              value: analytics?.topCategory ?? "—",
+              icon: <LayersIcon className="size-4" />,
+              iconClassName: "text-primary",
+            },
          ]}
        />
      </div>
@@ -312,58 +395,78 @@ export function FlowAnalyticsDetail({
        <Tabs value={slice} onValueChange={(v) => setSlice(String(v))} className="gap-3">
          <TabsList variant="line" className="flex flex-wrap h-auto">
            <TabsTrigger value="applications">Приложения</TabsTrigger>
+            <TabsTrigger value="categories">Категории</TabsTrigger>
+            <TabsTrigger value="services">Сервисы</TabsTrigger>
+            <TabsTrigger value="asns">ASN</TabsTrigger>
+            <TabsTrigger value="countries">Страны</TabsTrigger>
+            <TabsTrigger value="map">Карта</TabsTrigger>
            <TabsTrigger value="protocols">Протоколы</TabsTrigger>
            <TabsTrigger value="sources">Источники</TabsTrigger>
            <TabsTrigger value="destinations">Назначения</TabsTrigger>
-            <TabsTrigger value="conversations">Разговоры</TabsTrigger>
+            <TabsTrigger value="sessions">Сессии</TabsTrigger>
            <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 ?? []} onPick={(row) => pickBreakdown("category", row)} />
+          </TabsContent>
+          <TabsContent value="services">
+            <FlowBreakdownGrid rows={analytics?.services ?? []} onPick={(row) => pickBreakdown("service", row)} />
+          </TabsContent>
+          <TabsContent value="asns">
+            <FlowBreakdownGrid rows={analytics?.asns ?? []} onPick={(row) => pickBreakdown("asn", row)} />
+          </TabsContent>
+          <TabsContent value="countries">
+            <FlowBreakdownGrid rows={analytics?.countries ?? []} country onPick={(row) => pickBreakdown("country", row)} />
+          </TabsContent>
+          <TabsContent value="map">
+            <FlowTrafficMap
+              edges={analytics?.mapEdges ?? []}
+              onSelectCountry={(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="conversations">
+          <TabsContent value="sessions">
+            {sessionFilter ? (
+              <div className="flex items-center gap-2 mb-2">
+                <GlobeIcon className="size-3.5 text-muted-foreground" />
+                <Badge variant="secondary" size="sm">Фильтр: {sessionFilter.label}</Badge>
+                <button
+                  type="button"
+                  className="text-xs text-primary"
+                  onClick={() => setSessionFilter(null)}
+                >
+                  сбросить
+                </button>
+              </div>
+            ) : null}
            <TrafficFlowsDataGrid
-              rows={analytics?.conversationsList ?? []}
-              emptyHint={emptyHint}
+              rows={sessionRows}
+              emptyHint={emptyHint ?? "Нет сессий по выбранному фильтру"}
            />
          </TabsContent>
          <TabsContent value="interfaces">
-            <FlowBreakdownGrid rows={analytics?.interfaces ?? []} />
+            <FlowBreakdownGrid
+              rows={analytics?.interfaces ?? []}
+              empty="Нет данных по интерфейсам"
+              onPick={(row) => pickBreakdown("iface", row)}
+            />
          </TabsContent>
        </Tabs>
      </div>
-
-      <div className="mt-4 pt-4 border-t grid grid-cols-2 gap-4">
-        <div>
-          <p className="text-xs text-muted-foreground mb-1">Объём за период</p>
-          <p className="text-lg font-semibold tabular-nums">{formatBytes(bytes)}</p>
-          <Sparkline
-            data={analytics?.rxSeries ?? card.rxSeries}
-            width={180}
-            height={28}
-            color="var(--chart-rx)"
-            filled
-          />
-        </div>
-        <div>
-          <p className="text-xs text-muted-foreground mb-1">Уник. адреса</p>
-          <p className="text-lg font-semibold tabular-nums">
-            {analytics?.uniqueSrc ?? 0}
-            <span className="text-sm font-normal text-muted-foreground"> src · </span>
-            {analytics?.uniqueDst ?? 0}
-            <span className="text-sm font-normal text-muted-foreground"> dst</span>
-          </p>
-        </div>
-      </div>
    </>
  )
 }
@@ -0,0 +1,191 @@
+"use client"
+
+import { useMemo, useState } from "react"
+import { type ColumnDef, getCoreRowModel, useReactTable } from "@tanstack/react-table"
+import type { FlowMapEdge } from "@mmapp/contracts/traffic-flow"
+import { Frame, FramePanel } from "@/components/reui/frame"
+import { DataGridShell } from "@/components/data-grids/shared/data-grid-shell"
+import {
+  DATA_GRID_CELL_PAD,
+  DATA_GRID_CELL_PAD_FIRST,
+  DATA_GRID_CELL_PAD_LAST,
+} from "@/components/data-grids/shared/data-grid-layout"
+import { Flag } from "@/components/flag"
+import { fmtRate } from "@/lib/fmt-rate"
+
+const W = 640
+const H = 280
+const STROKES = [
+  "var(--chart-1)",
+  "var(--chart-2)",
+  "var(--chart-3)",
+  "var(--color-success)",
+  "var(--chart-5)",
+]
+
+function formatBytes(n: number): string {
+  if (n >= 1_000_000_000) return `${(n / 1_000_000_000).toFixed(2)} ГБ`
+  if (n >= 1_000_000) return `${(n / 1_000_000).toFixed(1)} МБ`
+  if (n >= 1000) return `${(n / 1000).toFixed(1)} КБ`
+  return `${n} Б`
+}
+
+function FlowTrafficMap({
+  edges,
+  onSelectCountry,
+}: {
+  edges: FlowMapEdge[]
+  onSelectCountry?: (country: string) => void
+}) {
+  const [hover, setHover] = useState<string | null>(null)
+
+  const layout = useMemo(() => {
+    const sources = [...new Map(edges.map((e) => [e.fromId, e])).values()]
+    const dests = [...new Map(edges.map((e) => [e.toCountry, e])).values()]
+    const srcY = (i: number) => sources.length <= 1 ? H / 2 : 36 + (i * (H - 72)) / Math.max(1, sources.length - 1)
+    const dstY = (i: number) => dests.length <= 1 ? H / 2 : 36 + (i * (H - 72)) / Math.max(1, dests.length - 1)
+    const srcPos = new Map(sources.map((s, i) => [s.fromId, { x: 88, y: srcY(i), label: s.fromLabel, country: s.fromCountry }]))
+    const dstPos = new Map(dests.map((d, i) => [d.toCountry, { x: 552, y: dstY(i), country: d.toCountry }]))
+    const maxBytes = Math.max(...edges.map((e) => e.bytes), 1)
+    return { srcPos, dstPos, maxBytes }
+  }, [edges])
+
+  const columns = useMemo<ColumnDef<FlowMapEdge>[]>(
+    () => [
+      {
+        id: "from",
+        accessorKey: "fromLabel",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Источник</span>,
+        cell: ({ row }) => (
+          <span className="flex items-center gap-1.5 text-sm">
+            {row.original.fromCountry && row.original.fromCountry !== "UN"
+              ? <Flag code={row.original.fromCountry} />
+              : null}
+            {row.original.fromLabel}
+          </span>
+        ),
+        meta: { headerClassName: DATA_GRID_CELL_PAD_FIRST, cellClassName: DATA_GRID_CELL_PAD_FIRST },
+      },
+      {
+        id: "to",
+        accessorKey: "toCountry",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Назначение</span>,
+        cell: ({ row }) => (
+          <span className="flex items-center gap-1.5 text-sm">
+            {/^[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>
+        ),
+        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
+      },
+      {
+        id: "cat",
+        accessorKey: "category",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Категория</span>,
+        cell: ({ row }) => <span className="text-xs">{row.original.category}</span>,
+        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
+      },
+      {
+        id: "rate",
+        accessorFn: (r) => r.bps,
+        header: () => <span className="text-xs font-medium text-muted-foreground">Скорость</span>,
+        cell: ({ row }) => <span className="text-xs tabular-nums">{fmtRate(row.original.bps / 1_000_000)}</span>,
+        meta: { headerClassName: DATA_GRID_CELL_PAD, cellClassName: DATA_GRID_CELL_PAD },
+      },
+      {
+        id: "bytes",
+        accessorKey: "bytes",
+        header: () => <span className="text-xs font-medium text-muted-foreground">Байты</span>,
+        cell: ({ row }) => <span className="text-xs tabular-nums">{formatBytes(row.original.bytes)}</span>,
+        meta: { headerClassName: DATA_GRID_CELL_PAD_LAST, cellClassName: DATA_GRID_CELL_PAD_LAST },
+      },
+    ],
+    [],
+  )
+
+  const table = useReactTable({
+    data: edges,
+    columns,
+    getCoreRowModel: getCoreRowModel(),
+    getRowId: (row, i) => `${row.fromId}-${row.toCountry}-${i}`,
+  })
+
+  if (!edges.length) {
+    return (
+      <p className="text-sm text-muted-foreground py-8 text-center">
+        Страны подтянутся из кэша RIPEstat
+      </p>
+    )
+  }
+
+  return (
+    <div className="flex flex-col gap-3">
+      <Frame>
+        <FramePanel className="p-3">
+          <svg viewBox={`0 0 ${W} ${H}`} className="w-full h-[220px]" role="img" aria-label="Карта потоков откуда куда">
+            {edges.map((e, i) => {
+              const from = layout.srcPos.get(e.fromId)
+              const to = layout.dstPos.get(e.toCountry)
+              if (!from || !to) return null
+              const id = `${e.fromId}-${e.toCountry}`
+              const midX = (from.x + to.x) / 2
+              const d = `M ${from.x} ${from.y} C ${midX} ${from.y}, ${midX} ${to.y}, ${to.x} ${to.y}`
+              const sw = 1.25 + 7 * (e.bytes / layout.maxBytes)
+              const active = hover === id
+              return (
+                <path
+                  key={id}
+                  d={d}
+                  fill="none"
+                  stroke={STROKES[i % STROKES.length]}
+                  strokeWidth={active ? sw + 1.5 : sw}
+                  strokeOpacity={active ? 1 : 0.72}
+                  className="cursor-pointer"
+                  onMouseEnter={() => setHover(id)}
+                  onMouseLeave={() => setHover(null)}
+                  onClick={() => onSelectCountry?.(e.toCountry)}
+                >
+                  <title>
+                    {`${e.fromLabel} → ${e.toCountry} · ${e.category} · ${formatBytes(e.bytes)}`}
+                  </title>
+                </path>
+              )
+            })}
+            {[...layout.srcPos.values()].map((n) => (
+              <g key={`s-${n.label}`}>
+                <circle cx={n.x} cy={n.y} r="7" className="fill-primary" />
+                <text x={n.x - 14} y={n.y + 4} textAnchor="end" className="fill-foreground text-[11px]">
+                  {n.label}
+                </text>
+              </g>
+            ))}
+            {[...layout.dstPos.values()].map((n) => (
+              <g key={`d-${n.country}`}>
+                <circle cx={n.x} cy={n.y} r="7" className="fill-chart-2" />
+                <text x={n.x + 14} y={n.y + 4} className="fill-foreground text-[11px]">
+                  {n.country}
+                </text>
+              </g>
+            ))}
+          </svg>
+          {hover ? (
+            <p className="text-[11px] text-muted-foreground mt-1">
+              Нажмите дугу, чтобы отфильтровать сессии по стране назначения
+            </p>
+          ) : null}
+        </FramePanel>
+      </Frame>
+      <DataGridShell
+        table={table}
+        recordCount={edges.length}
+        emptyMessage="Нет рёбер с известной страной"
+        onRowClick={(row) => onSelectCountry?.(row.toCountry)}
+      />
+    </div>
+  )
+}
+
+export { FlowTrafficMap }
@@ -22,6 +22,7 @@ export function useFlowLive(opts: {
  serverId?: string
  userId?: string
  iface?: string
+  dedup?: boolean
 }): { sample: FlowAnalyticsDto | null; error: string | null } {
  const [sample, setSample] = useState<FlowAnalyticsDto | null>(null)
  const [error, setError] = useState<string | null>(null)
@@ -41,6 +42,7 @@ export function useFlowLive(opts: {
      serverId: opts.serverId,
      userId: opts.userId,
      iface: opts.iface,
+      dedup: opts.dedup,
    })}`
    const url = resolveApiUrl(opts.backendUrl, path)

@@ -84,7 +86,7 @@ export function useFlowLive(opts: {
    })()

    return () => ac.abort()
-  }, [opts.enabled, opts.backendUrl, opts.range, opts.serverId, opts.userId, opts.iface])
+  }, [opts.enabled, opts.backendUrl, opts.range, opts.serverId, opts.userId, opts.iface, opts.dedup])

  return { sample, error }
 }
@@ -81,6 +81,10 @@ export const flowTalkerDtoSchema = z.object({
  inIface: z.string(),
  inIfaceIndex: z.string().optional(),
  application: z.string().optional(),
+  category: z.string().optional(),
+  service: z.string().optional(),
+  dstCountry: z.string().optional(),
+  dstAsn: z.number().int().optional(),
 })

 export const flowStatsDtoSchema = z.object({
@@ -133,14 +137,27 @@ export const flowEntityCardSchema = z.object({
  bytes: z.number().nonnegative(),
 })

+export const flowMapEdgeSchema = z.object({
+  fromId: z.string(),
+  fromLabel: z.string(),
+  fromCountry: z.string(),
+  toCountry: z.string(),
+  toAsn: z.number().int().nonnegative(),
+  category: z.string(),
+  bytes: z.number().nonnegative(),
+  bps: z.number().nonnegative(),
+})
+
 export const flowAnalyticsDtoSchema = z.object({
  bpsNow: z.number().nonnegative(),
  bytes: z.number().nonnegative(),
  packets: z.number().nonnegative(),
  conversations: z.number().int().nonnegative(),
+  conversationsRaw: z.number().int().nonnegative().optional(),
  uniqueSrc: z.number().int().nonnegative(),
  uniqueDst: z.number().int().nonnegative(),
  topProto: z.string(),
+  topCategory: z.string().optional(),
  rxSeries: z.array(z.number()),
  txSeries: z.array(z.number()),
  applications: z.array(flowBreakdownRowSchema),
@@ -148,9 +165,15 @@ export const flowAnalyticsDtoSchema = z.object({
  sources: z.array(flowBreakdownRowSchema),
  destinations: z.array(flowBreakdownRowSchema),
  interfaces: z.array(flowBreakdownRowSchema),
+  asns: z.array(flowBreakdownRowSchema).optional(),
+  countries: z.array(flowBreakdownRowSchema).optional(),
+  categories: z.array(flowBreakdownRowSchema).optional(),
+  services: z.array(flowBreakdownRowSchema).optional(),
+  mapEdges: z.array(flowMapEdgeSchema).optional(),
  conversationsList: z.array(flowTalkerDtoSchema),
  ifaces: z.array(flowIfaceChipSchema),
  live: z.boolean(),
+  dedupApplied: z.boolean().optional(),
 })

 export const flowExportersDtoSchema = z.object({
@@ -172,6 +195,7 @@ export type FlowStatsDto = z.infer<typeof flowStatsDtoSchema>
 export type FlowBreakdownRow = z.infer<typeof flowBreakdownRowSchema>
 export type FlowIfaceChip = z.infer<typeof flowIfaceChipSchema>
 export type FlowEntityCard = z.infer<typeof flowEntityCardSchema>
+export type FlowMapEdge = z.infer<typeof flowMapEdgeSchema>
 export type FlowAnalyticsDto = z.infer<typeof flowAnalyticsDtoSchema>
 export type FlowExportersDto = z.infer<typeof flowExportersDtoSchema>
 export type FlowClientsDto = z.infer<typeof flowClientsDtoSchema>
@@ -59,12 +59,15 @@ function flowQuery(params: {
  serverId?: string
  userId?: string
  iface?: string
+  dedup?: boolean
 }): string {
  const q = new URLSearchParams()
  if (params.range) q.set("range", params.range)
  if (params.serverId) q.set("serverId", params.serverId)
  if (params.userId) q.set("userId", params.userId)
  if (params.iface && params.iface !== "__all__") q.set("iface", params.iface)
+  if (params.dedup === false) q.set("dedup", "0")
+  else if (params.dedup === true) q.set("dedup", "1")
  const s = q.toString()
  return s ? `?${s}` : ""
 }
@@ -79,7 +82,7 @@ export async function getFlowClients(baseUrl: string, range = "5m"): Promise<Flo

 export async function getFlowAnalytics(
  baseUrl: string,
-  params: { range?: string; serverId?: string; userId?: string; iface?: string },
+  params: { range?: string; serverId?: string; userId?: string; iface?: string; dedup?: boolean },
 ): Promise<FlowAnalyticsDto> {
  return requestJson<FlowAnalyticsDto>(baseUrl, `/api/traffic/flow/analytics${flowQuery(params)}`)
 }
Author	SHA1	Message	Date
DenozordecandCursor	e0ddb17539	feat(traffic): классифицировать потоки по брендам и разгрузить ingest Docker images / prepare-release (push) Successful in 12s Details Docker images / backend-image (push) Successful in 1m48s Details Docker images / frontend-image (push) Successful in 2m53s Details Docker images / notify-webhook (push) Skipped Details Docker images / updater-image (push) Successful in 41s Details Docker images / publish-release (push) Successful in 8s Details Починить ISO-страну Cloudflare, не держать holder как сервис, клик по срезу открывает Сессии. Не вызывать REST /interface с UDP, flush SQLite пачкой — uptime не должен ловить timeout 10s. Co-authored-by: Cursor <[email protected]>	2026-09-07 03:01:14 +07:00
DenozordecandCursor	2820683cba	feat(traffic): добавить карту и срезы IPFIX по странам и сессиям Docker images / prepare-release (push) Successful in 11s Details Docker images / backend-image (push) Successful in 2m6s Details Docker images / frontend-image (push) Successful in 3m23s Details Docker images / notify-webhook (push) Skipped Details Docker images / updater-image (push) Successful in 49s Details Docker images / publish-release (push) Successful in 14s Details Имена ifIndex пишутся по обоим ключам REST, дедуп 5-tuple по max байт, RIPEstat только из prefix-кэша, карта откуда-куда в Frame. Co-authored-by: Cursor <[email protected]>	2026-09-07 02:13:21 +07:00