feat(statistics): enhance interface handling and data aggregation

Updated the statistics aggregation service to improve interface resolution and data handling. Introduced new functions for managing interface aliases and collapsing server interface rows, ensuring accurate data representation. Enhanced test coverage for interface resolution and added checks for new functionality. - Implemented `factIfaceAliases` and `collapseServerIfaceRows` for better interface data management. - Updated `resolveIfaceName` to handle additional cases for interface indexing. - Enhanced tests for interface resolution and aggregation logic. Co-authored-by: Cursor <[email protected]>
feat(network-map): enhance service selection logic and UI updates
2026-09-10 21:49:50 +07:00 · 2026-09-10 19:54:16 +07:00
12 changed files with 447 additions and 60 deletions
@@ -0,0 +1,63 @@
+# Локальные GeoLite2-базы (Country + ASN) для потоков по странам и ASN
+
+## Контекст
+
+Сейчас страна и ASN для netflow-потоков резолвятся через внешний RIPEstat API (`backend/src/services/traffic-flow-ripe.ts`): лимит 30 новых префиксов/мин, очередь на 90, кэш в PG `flow_ip_meta`. Новые IP «дозревают» с задержкой, IPv6 не покрывается (кэш индексируется только по IPv4). Локальные mmdb-базы дают мгновенный синхронный lookup всех IP без внешних вызовов.
+
+Решения (подтверждены):
+- Источник — **MaxMind GeoLite2 через P3TERX-зеркало**: `https://github.com/P3TERX/GeoLite.mmdb/raw/download/GeoLite2-Country.mmdb` и `.../GeoLite2-ASN.mmdb`. Без регистрации, ключей и tar-распаковки. Точность по стране у GeoLite2 и IPinfo паритетная (<1% ошибок у обоих, arXiv 2026); выбран P3TERX за надёжность зеркала (5.2k звёзд) и преемственность: текущий RIPE-путь и так читает GeoLite (`maxmind-geo-lite`), история в кэше остаётся консистентной.
+- **RIPEstat остаётся fallback** (до первой загрузки баз / если lookup не дал результата).
+- City-базу не качаем (lat/lng фронтенд не использует).
+
+## Изменения
+
+### 1. Зависимость
+- `npm install -w mikrotik-manager-backend maxmind` — sync-чтение mmdb, встроенные TS-типы, без транзитивных зависимостей, Node 22 ок.
+
+### 2. Новый сервис `backend/src/services/traffic-flow-geoip.ts`
+(по конвенциям окружения traffic-flow-*: контракт → маршрут → сервис, без БД-логики в маршрутах)
+- Каталог: `backend/storage/geoip/` (конвенция `storage/backups`), файлы `GeoLite2-Country.mmdb`, `GeoLite2-ASN.mmdb`.
+- `initGeoip()` — открыть ридеры best-effort при старте (из `index.ts` рядом с `startTrafficFlowListener`), независимо от настроек автообновления: файлы есть — работают.
+- `lookupGeoip(ip): FlowIpMeta | null` — синхронно: `country.iso_code` (fallback `registered_country.iso_code`) с валидацией `isIsoCountry`, ASN = `autonomous_system_number`, holder = `autonomous_system_organization`; приватные IP → negative-запись как в RIPE (`isNonPublicIp`); в PG не пишем (lookup и так быстрый). IPv6 поддержан ридером.
+- `resolveFlowIp(ip)` — фасад: `lookupGeoip(ip) ?? lookupRipeCached(ip)`; главный экспорт для потребителей.
+- `geoipStatus()` — loaded, даты сборки баз (метаданные mmdb). Тест-хук `setGeoipReadersForTests`. Смена ридеров после обновления — атомарная замена ссылок.
+
+### 3. Коллектор `backend/src/services/geoip-update-collector.ts`
+`collectGeoipUpdateOnce()` по образцу `certificate-renew-collector.ts`:
+1. Conditional GET с ETag/If-None-Match из настроек → 304 = skip (фолбэк-сравнение: размер/содержимое).
+2. Скачивание в `*.tmp` через глобальный `fetch` + AbortController с таймаутом (внешний HTTP из service-слоя — по правилу fastify-backend-drizzle).
+3. Валидация: открыть ридер из tmp-файла, пробой 8.8.8.8 (страна US, ASN 15169).
+4. `fs.rename` атомарная подмена, старый файл → `*.prev` (откат, если новый ридер не открылся).
+5. Перезагрузка ридеров, статус в настройках; snapshot для `scheduler_runs` (checked/downloaded/skipped/bytes/error).
+
+### 4. Планировщик (`backend/src/services/scheduler.ts`)
+- `JOB_KEYS` += `geoip_update`; case в `runSchedulerJobBody`; блок в `refreshScheduler()` по образцу `certificates_renew`: интервал `Math.max(6ч, updateIntervalSec*1000)`, по умолчанию 7 дней (upstream обновляется еженедельно) + немедленный первый запуск при включённой настройке.
+
+### 5. Схема и миграция
+- `backend/src/db/schema.ts`: singleton `geoip_settings` — `enabled` (default true), `updateIntervalSec` (default 604800), `lastCheckAt`, `lastSuccessAt`, `lastError`, `countryBuildAt`, `asnBuildAt`, `etagsJson` (jsonb), `createdAt/updatedAt`.
+- Миграция: `npm run db:generate` → файл в `backend/drizzle/`.
+
+### 6. API + контракты
+- `packages/contracts/src/geoip.ts`: zod-схемы настроек/статуса (все входы — Zod, по правилам проекта).
+- Новый `backend/src/routes/geoip.ts`, регистрация в `index.ts` с prefix `/api`:
+  - `GET /api/geoip` — настройки + статус (ready, даты сборки, последняя проверка/ошибка);
+  - `PUT /api/geoip` — сохранить настройки, затем `refreshScheduler()`;
+  - `POST /api/geoip/update` — запустить загрузку сейчас (409, если уже идёт; флаг-гард как в коллекторах).
+
+### 7. Интеграция в пайплайн (geoip-first, RIPE-fallback)
+- `traffic-flow-engine.ts` (`queueParsedFlows`, ~строка 336): `lookupRipeCached` → `resolveFlowIp`. Логика misses не меняется: при готовом mmdb публичные IP (v4+v6) резолвятся сразу, очередь RIPE пустеет; до скачивания баз — прежнее поведение.
+- Остальные вызовы `lookupRipeCached` → `resolveFlowIp` (grep: как минимум `traffic-flow-analytics.ts` ~258–271).
+- `classifyFlowDst`/бренды не трогаем: holder из mmdb (org name) встаёт в существующие `HOLDER_BRANDS`-регулярки как есть.
+
+### 8. Frontend (по next-shadcn-production / ui-guardian: только переиспользование)
+- Секция «GeoIP-базы (GeoLite2)» внутри существующей `components/traffic/netflow-settings-panel.tsx`: статус (готово/не скачано, даты сборки Country/ASN, последняя проверка, ошибка), тумблер автообновления, интервал, кнопка «Обновить сейчас» с индикатором. Только уже используемые в панели примитивы (Switch/Button/поля) — никаких новых визуальных паттернов и Card-shell. API-клиент через существующие http-хелперы.
+
+### 9. Хаускипинг, тесты, проверки
+- `backend/.gitignore`: `storage/geoip/`.
+- Тесты `backend/src/services/traffic-flow-geoip.test.ts` + скрипт `test:geoip` (по образцу `test:traffic-flow`): приоритет фасада (geoip hit → RIPE не зовётся; miss → fallback), negative на приватных IP, фильтрация EU/ZZ через `isIsoCountry`, коллектор с мокнутым fetch (304-skip, битый файл → подмены нет, `.prev` сохранён), dims по стране/ASN с засеянным ридером.
+- Проверки после реализации (обязательно по правилам): типы/сборка бэка (`npm run build -w mikrotik-manager-backend`), типы фронта при правке UI (`npx tsc --noEmit`), `npm run test:geoip` и `test:traffic-flow`; предупреждения не игнорировать.
+- Коммит: `feat(netflow): <subject по-русски>` — новая пользовательская фича (мгновенные страна/ASN в потоках), по commit-messages-ru.
+- README: короткий раздел о GeoIP; примечание, что в Docker `storage/geoip` ephemeral без тома — базы перекачаются после пересоздания контейнера (~17 МБ); при желании смонтировать volume.
+
+## Что это даёт
+Страна и ASN появляются у потока мгновенно при ingest (включая IPv6), без ограничения скорости RIPE; dims `country`/`asn` в `flow_daily_dims`, аналитика (карта, топы, monthly) становятся полными сразу. Внешняя зависимость от stat.ripe.net остаётся только как fallback до первой загрузки баз.
@@ -1216,6 +1216,9 @@ export default function NetworkMapPage() {
  // ── Interaction ─────────────────────────────────────────────────────────────
  const [selected,   setSelected]   = useState<Server | null>(null)
  const [selectedService, setSelectedService] = useState<FlowMapService | null>(null)
+  const liveSelectedService = selectedService
+    ? (mapServices.find((s) => s.id === selectedService.id) ?? selectedService)
+    : null
  const [highlightedPath, setHighlightedPath] = useState<{ viaId: string; enId: string; serviceId: string } | null>(null)
  const [selWanIdx,  setSelWanIdx]  = useState<number | null>(null)
  const [hoveredId,  setHoveredId]  = useState<string | null>(null)
@@ -2702,16 +2705,16 @@ export default function NetworkMapPage() {
                  })()}
                </div>
              </>
-            ) : selectedService ? (
+            ) : liveSelectedService ? (
              <>
                <div className="flex items-start gap-2 px-4 py-3 border-b">
                  <div className="mt-0.5">
-                    <ServiceBrandIcon label={selectedService.label} size={22} />
+                    <ServiceBrandIcon label={liveSelectedService.label} size={22} />
                  </div>
                  <div className="flex-1 min-w-0">
-                    <p className="font-mono font-semibold text-sm truncate">{selectedService.label}</p>
+                    <p className="font-mono font-semibold text-sm truncate">{liveSelectedService.label}</p>
                    <p className="text-xs text-muted-foreground mt-0.5">
-                      Конечный сервис · {selectedService.category}
+                      Конечный сервис · {liveSelectedService.category}
                    </p>
                  </div>
                  <button
@@ -2726,15 +2729,15 @@ export default function NetworkMapPage() {
                  <div className="flex flex-col gap-0">
                    <div className="flex items-center justify-between py-2 border-b border-border/50">
                      <span className="text-xs text-muted-foreground">Доля окна</span>
-                      <span className="text-xs font-mono font-medium text-cyan-400">{serviceSharePct(selectedService.share)}</span>
+                      <span className="text-xs font-mono font-medium text-cyan-400">{serviceSharePct(liveSelectedService.share)}</span>
                    </div>
                    <div className="flex items-center justify-between py-2 border-b border-border/50">
                      <span className="text-xs text-muted-foreground">Скорость</span>
                      <span className="text-xs font-mono font-medium">
                        {formatNetflowRate({
-                          bytes: selectedService.bytes,
-                          bps: selectedService.bps,
-                          bpsFwd: selectedService.bps,
+                          bytes: liveSelectedService.bytes,
+                          bps: liveSelectedService.bps,
+                          bpsFwd: liveSelectedService.bps,
                          bpsRev: 0,
                        })}
                      </span>
@@ -2742,34 +2745,34 @@ export default function NetworkMapPage() {
                  </div>
                  <div>
                    <p className="text-xs font-semibold text-muted-foreground uppercase tracking-wider mb-2">Выход</p>
-                    <div className="flex flex-col gap-1.5">
-                      {visibleServiceEdges.filter((e) => e.toId === selectedService.id).map((e) => {
+                    <div className="flex flex-col gap-3">
+                      {visibleServiceEdges.filter((e) => e.toId === liveSelectedService.id).map((e) => {
                        const src = mapServers.find((s) => s.id === e.fromId)
+                        const enPaths = mapServicePaths
+                          .filter((p) => p.serviceId === liveSelectedService.id && p.enId === e.fromId)
+                          .slice()
+                          .sort((a, b) => b.bps - a.bps)
                        return (
-                          <div key={`${e.fromId}|${e.toId}`} className="flex items-center justify-between text-xs">
-                            <span className="font-mono truncate">{src?.name ?? e.fromId}</span>
-                            <span className="font-mono text-emerald-400 tabular-nums">
-                              {formatNetflowRate({ bytes: e.bytes, bps: e.bps, bpsFwd: e.bpsFwd, bpsRev: e.bpsRev })}
-                            </span>
+                          <div key={`${e.fromId}|${e.toId}`} className="flex flex-col gap-1.5">
+                            <div className="flex items-center justify-between text-xs">
+                              <span className="font-mono truncate">{src?.name ?? e.fromId}</span>
+                              <span className="font-mono text-emerald-400 tabular-nums">
+                                {formatNetflowRate({ bytes: e.bytes, bps: e.bps, bpsFwd: e.bpsFwd, bpsRev: e.bpsRev })}
+                              </span>
+                            </div>
+                            <ServicePathList
+                              paths={enPaths}
+                              servers={mapServers}
+                              services={mapServices}
+                              highlight={highlightedPath}
+                              viaMode="via"
+                              onToggle={togglePathHighlight}
+                            />
                          </div>
                        )
                      })}
                    </div>
                  </div>
-                  <div>
-                    <p className="text-xs font-semibold text-muted-foreground uppercase tracking-wider mb-2">Пути</p>
-                    <ServicePathList
-                      paths={mapServicePaths
-                        .filter((p) => p.serviceId === selectedService.id)
-                        .slice()
-                        .sort((a, b) => b.bps - a.bps)}
-                      servers={mapServers}
-                      services={mapServices}
-                      highlight={highlightedPath}
-                      viaMode="via"
-                      onToggle={togglePathHighlight}
-                    />
-                  </div>
                </div>
              </>
            ) : selected ? (
@@ -3002,7 +3005,11 @@ export default function NetworkMapPage() {
                <p className="text-xs font-semibold text-muted-foreground uppercase tracking-wider mb-2">Пути</p>
                <ServicePathList
                  paths={mapServicePaths
-                    .filter((p) => p.viaId === selected.id || p.enId === selected.id)
+                    .filter((p) => (
+                      selected.type === "exit-node"
+                        ? p.enId === selected.id
+                        : p.viaId === selected.id
+                    ))
                    .slice()
                    .sort((a, b) => b.bps - a.bps)}
                  servers={mapServers}
@@ -1,6 +1,7 @@
 import assert from "node:assert/strict"
 import { getStatistics, getStatisticsPivot, parseStatisticsPeriod, pivotDimsConflict } from "./statistics-aggregate.js"
 import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js"
+import { setRefreshIfacesForTests } from "./traffic-flow-ifaces.js"
 import { withPgOrSkip } from "../test/pg.js"
 import { dbQuery } from "../db/index.js"
 import { ensurePartitionFor } from "../db/partitions.js"
@@ -52,6 +53,7 @@ await dbQuery(`

 resetIfaceCacheForTests()
 rememberServerIfaces(serverId, [{ name: "gre-client", ifindex: "2" }])
+setRefreshIfacesForTests(async () => {})

 await dbQuery(`
  INSERT INTO flow_daily_facts (server_id, day, iface, country, service, asn, bytes, packets)
@@ -72,6 +74,11 @@ try {
  assert.ok(unbound)
  assert.equal(unbound.bytes, 70)
  assert.ok(all.servers.some((r) => r.id === String(serverId)))
+  const greIface = all.interfaces.find((r) => r.label.includes("gre-client"))
+  assert.ok(greIface)
+  assert.equal(greIface.bytes, 1000)
+  assert.equal(greIface.id, `${serverId}:gre-client`)
+  assert.ok(!all.interfaces.some((r) => /· (?:#)?\d+$/.test(r.label)))

  const sliced = await getStatistics({
    from: "2026-09-01",
@@ -118,6 +125,7 @@ try {
  assert.equal(hourly.kpis.bytes, 40)
  assert.ok(hourly.users.some((r) => r.id === "u-stats-1"))
 } finally {
+  setRefreshIfacesForTests(null)
  resetIfaceCacheForTests()
  await dbQuery(`DELETE FROM flow_daily_facts WHERE server_id = $1`, [serverId])
  await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id = $1`, [serverId])
@@ -11,10 +11,12 @@ import {
  type StatisticsQuery,
 } from "@mmapp/contracts/statistics"
 import {
-  bindingIfaceAliases,
-  bindingIfaceAliasesAllServers,
+  collapseServerIfaceRows,
+  displayFactIface,
  expandBindingIfaces,
+  factIfaceAliases,
 } from "./traffic-flow-ifindex.js"
+import { refreshServerIfaces } from "./traffic-flow-ifaces.js"

 const TOP_N = 200
 const HOUR_WINDOW_MS = 48 * 3600_000
@@ -92,10 +94,31 @@ interface FilterCtx {
 }

 function ifaceFilterAliases(iface: string, serverId?: number): string[] {
+  return factIfaceAliases(iface.trim(), serverId)
+}
+
+function looksLikeIfIndex(iface: string): boolean {
  const raw = iface.trim()
-  if (!raw) return []
-  if (serverId != null) return bindingIfaceAliases(serverId, raw)
-  return bindingIfaceAliasesAllServers(raw)
+  return /^\d+$/.test(raw) || /^#\d+$/.test(raw)
+}
+
+async function warmIfaceCache(ids: Iterable<number>): Promise<void> {
+  const uniq = [...new Set(ids)].filter((id) => Number.isFinite(id) && id > 0)
+  if (!uniq.length) return
+  await Promise.all(uniq.map((id) => refreshServerIfaces(id)))
+}
+
+async function warmBindingIfaceCache(): Promise<void> {
+  const rows = await db.select({ serverId: userInterfaceBindings.serverId }).from(userInterfaceBindings)
+  await warmIfaceCache(rows.map((r) => r.serverId))
+}
+
+function canonicalIfaceDimId(id: string): string {
+  const colon = id.indexOf(":")
+  if (colon < 0) return id
+  const sid = Number(id.slice(0, colon))
+  if (!Number.isFinite(sid)) return id
+  return `${sid}:${displayFactIface(sid, id.slice(colon + 1))}`
 }

 function factWhere(alias: string, grain: "hour" | "day", ctx: FilterCtx): { sql: string; params: unknown[] } {
@@ -214,7 +237,7 @@ async function loadBindUserTuples(): Promise<UserBindTuple[]> {
  const seen = new Set<string>()
  const out: UserBindTuple[] = []
  for (const b of binds) {
-    for (const iface of bindingIfaceAliases(b.serverId, b.interfaceName)) {
+    for (const iface of factIfaceAliases(b.interfaceName, b.serverId)) {
      const k = `${b.userId}\0${b.serverId}\0${iface}`
      if (seen.has(k)) continue
      seen.add(k)
@@ -281,6 +304,8 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
    windowSec: 1,
  })

+  await warmBindingIfaceCache()
+  if (query.serverId) await warmIfaceCache([query.serverId])
  const bindTuples = await loadBindUserTuples()
  const ctx = await buildFilterCtx(query, period)
  if (!ctx) return emptyDto(period)
@@ -289,12 +314,11 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
  const timeCol = period.grain === "hour" ? "bucket_at" : "day"
  const where = factWhere("f", period.grain, ctx)

-  const totals = await dbAll<{ bytes: number; packets: number; servers: number; ifaces: number }>(`
+  const totals = await dbAll<{ bytes: number; packets: number; servers: number }>(`
    SELECT
      COALESCE(SUM(f.bytes), 0) AS bytes,
      COALESCE(SUM(f.packets), 0) AS packets,
-      COUNT(DISTINCT f.server_id)::int AS servers,
-      COUNT(DISTINCT (f.server_id::text || ':' || f.iface))::int AS ifaces
+      COUNT(DISTINCT f.server_id)::int AS servers
    FROM ${table} f
    WHERE ${where.sql}
  `, where.params)
@@ -302,7 +326,6 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
  const bytes = Number(totals[0]?.bytes) || 0
  const packets = Number(totals[0]?.packets) || 0
  const serverCount = Number(totals[0]?.servers) || 0
-  const ifaceCount = Number(totals[0]?.ifaces) || 0

  const seriesRows = await dbAll<{ t: string; bytes: number }>(`
    SELECT ${timeCol}::text AS t, SUM(f.bytes) AS bytes
@@ -340,12 +363,15 @@ export async function getStatistics(query: StatisticsQuery): Promise<StatisticsD
    GROUP BY f.server_id
  `, where.params)

-  const ifaceRows = await dbAll<{ serverId: number; iface: string; bytes: number; packets: number }>(`
+  const ifaceRowsRaw = await dbAll<{ serverId: number; iface: string; bytes: number; packets: number }>(`
    SELECT f.server_id AS "serverId", f.iface AS iface, SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
    FROM ${table} f
    WHERE ${where.sql}
    GROUP BY f.server_id, f.iface
  `, where.params)
+  await warmIfaceCache(ifaceRowsRaw.filter((r) => looksLikeIfIndex(r.iface)).map((r) => r.serverId))
+  const ifaceRows = collapseServerIfaceRows(ifaceRowsRaw)
+  const ifaceCount = ifaceRows.length

  let userRows: Array<{ id: string; bytes: number; packets: number }> = []
  if (bindTuples.length && !ctx.unboundOnly) {
@@ -522,6 +548,8 @@ export async function getStatisticsPivot(query: StatisticsPivotQuery): Promise<S
  if (pivotDimsConflict(query.row, query.col)) return emptyPivot(query)
  const period = parseStatisticsPeriod(query.from, query.to)
  if (!period) return emptyPivot(query)
+  await warmBindingIfaceCache()
+  if (query.serverId) await warmIfaceCache([query.serverId])
  const bindTuples = await loadBindUserTuples()
  const ctx = await buildFilterCtx(query, period)
  if (!ctx) return emptyPivot(query)
@@ -543,6 +571,25 @@ export async function getStatisticsPivot(query: StatisticsPivotQuery): Promise<S
    GROUP BY 1, 2
  `, [...join.params, ...where.params])

+  if (query.row === "iface" || query.col === "iface") {
+    const ifaceServerIds: number[] = []
+    for (const r of raw) {
+      for (const dim of [query.row, query.col] as const) {
+        if (dim !== "iface") continue
+        const id = dim === query.row ? String(r.row_id ?? "") : String(r.col_id ?? "")
+        const colon = id.indexOf(":")
+        if (colon < 0) continue
+        const sid = Number(id.slice(0, colon))
+        if (looksLikeIfIndex(id.slice(colon + 1)) && Number.isFinite(sid)) ifaceServerIds.push(sid)
+      }
+    }
+    await warmIfaceCache(ifaceServerIds)
+    for (const r of raw) {
+      if (query.row === "iface") r.row_id = canonicalIfaceDimId(String(r.row_id ?? ""))
+      if (query.col === "iface") r.col_id = canonicalIfaceDimId(String(r.col_id ?? ""))
+    }
+  }
+
  const metric = query.metric
  type Acc = { bytes: number; packets: number }
  const cell = new Map<string, Map<string, Acc>>()
@@ -684,7 +731,9 @@ async function loadPivotLabels(
      if (colon < 0) return id
      const sid = id.slice(0, colon)
      const iface = id.slice(colon + 1)
-      return `${serverNames.get(sid) || sid} · ${iface}`
+      const sidNum = Number(sid)
+      const name = Number.isFinite(sidNum) ? displayFactIface(sidNum, iface) : iface
+      return `${serverNames.get(sid) || sid} · ${name}`
    }
    return id
  }
@@ -1,5 +1,10 @@
 import assert from "node:assert/strict"
-import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js"
+import {
+  dedupFlowRowsAcrossExporters,
+  dedupFlowRowsMaxBytes,
+  flowConversationKey,
+  flowTupleKey,
+} from "./traffic-flow-dedup.js"

 const a = {
  serverId: 7,
@@ -22,4 +27,13 @@ assert.equal(flowTupleKey(a), flowTupleKey(b))
 const sameIface = dedupFlowRowsMaxBytes([a, { ...a, bytes: 3_000, packets: 2 }])
 assert.equal(sameIface[0]?.bytes, 15_000)

+const jh = { ...a, serverId: 7, bytes: 9_000 }
+const en = { ...a, serverId: 9, bytes: 11_000, inIface: "1" }
+assert.equal(flowConversationKey(jh), flowConversationKey(en))
+assert.notEqual(flowTupleKey(jh), flowTupleKey(en))
+const across = dedupFlowRowsAcrossExporters([en, jh], (x, y) => (x.serverId === 7 ? x : y))
+assert.equal(across.length, 1)
+assert.equal(across[0]?.serverId, 7)
+assert.equal(across[0]?.bytes, 9_000)
+
 console.log("traffic-flow-dedup.test.ts: ok")
@@ -14,6 +14,32 @@ export function flowTupleKey(r: Pick<FlowTupleRow, "serverId" | "src" | "dst" |
  return `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}`
 }

+/** Один разговор на всех экспортёрах (JH+EN), без serverId. */
+export function flowConversationKey(r: Pick<FlowTupleRow, "src" | "dst" | "proto" | "srcPort" | "dstPort">): string {
+  return `${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}`
+}
+
+/**
+ * Схлопнуть копии одного 5-tuple с разных серверов.
+ * `prefer` выбирает ряд (клиент на JH важнее голого EN).
+ */
+export function dedupFlowRowsAcrossExporters<T extends FlowTupleRow>(
+  rows: T[],
+  prefer: (a: T, b: T) => T,
+): T[] {
+  const byConv = new Map<string, T>()
+  for (const row of rows) {
+    const key = flowConversationKey(row)
+    const prev = byConv.get(key)
+    if (!prev) {
+      byConv.set(key, row)
+      continue
+    }
+    byConv.set(key, prefer(prev, row))
+  }
+  return [...byConv.values()]
+}
+
 function ifaceKey(r: FlowTupleRow): string {
  return `${flowTupleKey(r)}|${r.inIface}`
 }
@@ -22,8 +22,9 @@ rememberServerIfaces(7, [
  { ".id": "*A", name: "wg-flow" },
  { ".id": "*D", name: "bridge" },
 ])
-assert.equal(resolveIfaceName(7, "2").name, "ether1")
-assert.equal(resolveIfaceName(7, "10").name, "wg-flow")
+  assert.equal(resolveIfaceName(7, "2").name, "ether1")
+  assert.equal(resolveIfaceName(7, "#2").name, "ether1")
+  assert.equal(resolveIfaceName(7, "10").name, "wg-flow")
 assert.equal(resolveIfaceName(7, "13").name, "bridge")
 assert.equal(resolveIfaceName(7, "0").name, "—")
 assert.equal(resolveIfaceName(7, "ether1").name, "ether1")
@@ -3,7 +3,10 @@ import {
  bindingIfaceAliases,
  bindingIfaceAliasesAllServers,
  canonicalFactIface,
+  collapseServerIfaceRows,
+  displayFactIface,
  expandBindingIfaces,
+  factIfaceAliases,
  rememberServerIfaces,
  resetIfaceCacheForTests,
  resolveIfaceName,
@@ -18,12 +21,20 @@ assert.equal(canonicalFactIface(1, "2"), "gre-client")
 assert.equal(canonicalFactIface(1, "gre-client"), "gre-client")
 assert.equal(canonicalFactIface(1, "9"), "9")
 assert.equal(resolveIfaceName(1, "9").name, "#9")
+assert.equal(resolveIfaceName(1, "2").name, "gre-client")
+assert.equal(resolveIfaceName(1, "#2").name, "gre-client")
+assert.equal(displayFactIface(1, "2"), "gre-client")

 const aliases = bindingIfaceAliases(1, "gre-client")
 assert.ok(aliases.includes("gre-client"))
 assert.ok(aliases.includes("2"))
 assert.ok(aliases.includes("#2"))

+const fromIndex = factIfaceAliases("2", 1)
+assert.ok(fromIndex.includes("gre-client"))
+assert.ok(fromIndex.includes("2"))
+assert.ok(fromIndex.includes("#2"))
+
 const all = bindingIfaceAliasesAllServers("gre-client")
 assert.ok(all.includes("2"))

@@ -31,5 +42,17 @@ const expanded = expandBindingIfaces([{ serverId: 1, iface: "gre-client" }])
 assert.ok(expanded.some((x) => x.iface === "2"))
 assert.ok(expanded.some((x) => x.iface === "gre-client"))

+const collapsed = collapseServerIfaceRows([
+  { serverId: 1, iface: "2", bytes: 10, packets: 1 },
+  { serverId: 1, iface: "gre-client", bytes: 5, packets: 2 },
+  { serverId: 1, iface: "wan1", bytes: 3, packets: 1 },
+])
+assert.equal(collapsed.length, 2)
+const gre = collapsed.find((r) => r.iface === "gre-client")
+assert.ok(gre)
+assert.equal(gre.bytes, 15)
+assert.equal(gre.packets, 3)
+assert.ok(collapsed.some((r) => r.iface === "wan1"))
+
 resetIfaceCacheForTests()
 console.log("traffic-flow-ifindex.test.ts: ok")
@@ -36,12 +36,12 @@ export function rememberServerIfaces(serverId: number, rows: RosIfaceIndexRow[])

 export function resolveIfaceName(serverId: number, indexOrName: string): { name: string; index: string } {
  const trimmed = String(indexOrName ?? "").trim()
-  if (!trimmed || trimmed === "0") return { name: "—", index: trimmed }
-  if (!/^\d+$/.test(trimmed)) return { name: trimmed, index: "" }
-  const idx = Number(trimmed)
-  const name = cache.get(serverId)?.get(idx)
-  if (name) return { name, index: trimmed }
-  return { name: `#${trimmed}`, index: trimmed }
+  const asIndex = trimmed.startsWith("#") && /^\d+$/.test(trimmed.slice(1)) ? trimmed.slice(1) : trimmed
+  if (!asIndex || asIndex === "0") return { name: "—", index: asIndex }
+  if (!/^\d+$/.test(asIndex)) return { name: trimmed, index: "" }
+  const name = cache.get(serverId)?.get(Number(asIndex))
+  if (name) return { name, index: asIndex }
+  return { name: `#${asIndex}`, index: asIndex }
 }

 /** Имя iface для факта куба: ifIndex→имя, без `#13` при пустом кэше. */
@@ -53,18 +53,86 @@ export function canonicalFactIface(serverId: number, inIface: string): string {
  return name || trimmed
 }

+function numericIfaceIndex(iface: string): string | null {
+  const raw = String(iface ?? "").trim()
+  if (/^\d+$/.test(raw)) return raw
+  if (raw.startsWith("#") && /^\d+$/.test(raw.slice(1))) return raw.slice(1)
+  return null
+}
+
+/** Имя для UI: ifIndex → RouterOS name; `0` → «—»; miss → `#n`. */
+export function displayFactIface(serverId: number, iface: string): string {
+  return resolveIfaceName(serverId, iface).name
+}
+
+/** Склеить факты `2` + `ether1` в одну строку после резолва ifIndex. */
+export function collapseServerIfaceRows(
+  rows: Array<{ serverId: number; iface: string; bytes: number; packets: number }>,
+): Array<{ serverId: number; iface: string; bytes: number; packets: number }> {
+  const acc = new Map<string, { serverId: number; iface: string; bytes: number; packets: number }>()
+  for (const r of rows) {
+    const name = displayFactIface(r.serverId, r.iface)
+    const k = `${r.serverId}\0${name}`
+    const prev = acc.get(k)
+    const bytes = Number(r.bytes) || 0
+    const packets = Number(r.packets) || 0
+    if (prev) {
+      prev.bytes += bytes
+      prev.packets += packets
+    } else {
+      acc.set(k, { serverId: r.serverId, iface: name, bytes, packets })
+    }
+  }
+  return [...acc.values()]
+}
+
+/** Ключи факта для фильтра: имя, ifIndex и `#n`. */
+export function factIfaceAliases(iface: string, serverId?: number): string[] {
+  const raw = String(iface ?? "").trim()
+  if (!raw) return []
+  const out = new Set<string>([raw])
+  const idx = numericIfaceIndex(raw)
+  if (idx) {
+    out.add(idx)
+    out.add(`#${idx}`)
+    const n = Number(idx)
+    if (serverId != null) {
+      const name = cache.get(serverId)?.get(n)
+      if (name) out.add(name)
+    } else {
+      for (const map of cache.values()) {
+        const name = map.get(n)
+        if (name) out.add(name)
+      }
+    }
+  }
+  if (serverId != null) {
+    for (const a of bindingIfaceAliases(serverId, raw)) out.add(a)
+  } else {
+    for (const a of bindingIfaceAliasesAllServers(raw)) out.add(a)
+  }
+  return [...out]
+}
+
 /** Имя + ifIndex + `#n` — тот же матч, что карта `/traffic`. */
 export function bindingIfaceAliases(serverId: number, interfaceName: string): string[] {
  const name = String(interfaceName ?? "").trim()
  if (!name) return []
  const out = new Set<string>([name])
  const map = cache.get(serverId)
-  if (!map) return [...out]
-  for (const [idx, n] of map) {
-    if (n !== name) continue
-    out.add(String(idx))
+  const idx = numericIfaceIndex(name)
+  const canonical = (idx && map?.get(Number(idx))) || name
+  out.add(canonical)
+  if (idx) {
+    out.add(idx)
    out.add(`#${idx}`)
  }
+  if (!map) return [...out]
+  for (const [i, n] of map) {
+    if (n !== canonical && n !== name) continue
+    out.add(String(i))
+    out.add(`#${i}`)
+  }
  return [...out]
 }

@@ -584,4 +584,98 @@ try {
  resetFlowCatalogForTests()
 }

+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+seedFlowTopologyForTests(topo)
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "gre-client" },
+  { ".id": "*3", name: "gre-jh-en" },
+])
+rememberServerIfaces(9, [
+  { ".id": "*1", name: "ether1" },
+])
+googleRipe()
+ingestParsedFlowsForServerForTests(7, [payloadFlow("8.8.8.8", 12_000)])
+ingestParsedFlowsForServerForTests(9, [{
+  src: "10.100.1.17",
+  dst: "8.8.8.8",
+  proto: 6,
+  srcPort: 51234,
+  dstPort: 443,
+  bytes: 12_000,
+  packets: 10,
+  inIface: "1",
+  outIface: "1",
+  nextHop: "",
+}])
+try {
+  resetFlowMapHopsCacheForTests()
+  const dual = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
+  const gre = dual.hops.find((h) => h.kind === "gre" && h.fromId === "7" && h.toId === "9")
+  assert.ok(gre, "GRE JH→EN сохранён")
+  assert.equal(gre.bytes, 12_000)
+  const googlePaths = (dual.servicePaths ?? []).filter((p) => p.serviceId === "svc:google")
+  assert.equal(googlePaths.length, 1, "один путь без копии EN")
+  assert.equal(googlePaths[0]?.clientId, "u1")
+  assert.equal(googlePaths[0]?.viaId, "7")
+  const googleEdge = dual.serviceEdges?.find((e) => e.toId === "svc:google" && e.fromId === "9")
+  assert.ok(googleEdge)
+  assert.equal(googleEdge.bytes, 12_000)
+  assert.equal(googleEdge.bps, googlePaths[0]?.bps)
+} finally {
+  seedFlowTopologyForTests(null)
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+seedFlowTopologyForTests(topo)
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "gre-client" },
+  { ".id": "*3", name: "gre-jh-en" },
+])
+googleRipe()
+ingestParsedFlowsForServerForTests(7, [
+  payloadFlow("8.8.8.8", 8_000),
+  {
+    src: "8.8.8.8",
+    dst: "10.100.1.17",
+    proto: 6,
+    srcPort: 443,
+    dstPort: 51234,
+    bytes: 4_000,
+    packets: 8,
+    inIface: "3",
+    outIface: "2",
+    nextHop: "",
+  },
+])
+try {
+  resetFlowMapHopsCacheForTests()
+  const bothDir = await buildFlowMapHops({ minutes: 5, minSharePct: 0 })
+  const gre = bothDir.hops.find((h) => h.kind === "gre" && h.fromId === "7" && h.toId === "9")
+  assert.ok(gre, "GRE-hop при fwd/rev")
+  const googlePaths = (bothDir.servicePaths ?? []).filter((p) => p.serviceId === "svc:google")
+  assert.equal(googlePaths.length, 1, "fwd+rev — один клиент")
+  assert.equal(googlePaths[0]?.clientId, "u1")
+  assert.ok(!(bothDir.servicePaths ?? []).some((p) => p.serviceId === "svc:google" && p.clientId === "—"))
+  const googleEdge = bothDir.serviceEdges?.find((e) => e.toId === "svc:google" && e.fromId === "9")
+  assert.ok(googleEdge)
+  assert.equal(googleEdge.bytes, 12_000)
+  assert.equal(googleEdge.bps, googlePaths[0]?.bps)
+} finally {
+  seedFlowTopologyForTests(null)
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
 console.log("traffic-flow-map-hops.test.ts: ok")
@@ -8,7 +8,7 @@ import {
  mapServiceNodeId,
  resolveFlowBrand,
 } from "./traffic-flow-brands.js"
-import { dedupFlowRowsMaxBytes } from "./traffic-flow-dedup.js"
+import { dedupFlowRowsAcrossExporters, dedupFlowRowsMaxBytes } from "./traffic-flow-dedup.js"
 import { getFlowListenerState, listFlowRowsForWindow } from "./traffic-flow-ingest.js"
 import { resolveIfaceName } from "./traffic-flow-ifaces.js"
 import { classifyFlowPlane, shouldKeepPlane } from "./traffic-flow-planes.js"
@@ -16,7 +16,7 @@ import { pickInternetPeer } from "./traffic-flow-ip.js"
 import { type FlowIpMeta } from "./traffic-flow-ripe.js"
 import { resolveFlowIp } from "./traffic-flow-geoip.js"
 import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js"
-import { loadFlowTopology, resolveClient, resolveEn, getServerCatalog } from "./traffic-flow-topology.js"
+import { loadFlowTopology, resolveClient, resolveEn, getServerCatalog, type FlowTopology } from "./traffic-flow-topology.js"
 import { flowDataEpoch } from "./traffic-flow-engine.js"

 export const DEFAULT_MAP_SERVICE_MIN_SHARE_PCT = 5
@@ -141,6 +141,16 @@ function ifaceUsable(name: string): boolean {
  return Boolean(name) && name !== "—"
 }

+function resolveMapClient(
+  topo: FlowTopology,
+  serverId: number,
+  inName: string,
+  outName: string,
+) {
+  return resolveClient(topo, serverId, inName)
+    ?? (ifaceUsable(outName) ? resolveClient(topo, serverId, outName) : null)
+}
+
 function bump(acc: Map<string, HopAcc>, key: string, seed: Omit<HopAcc, "bytes" | "bytesFwd" | "bytesRev">, bytes: number, dir: "fwd" | "rev" | "both"): void {
  const prev = acc.get(key)
  const addFwd = dir === "fwd" || dir === "both" ? bytes : 0
@@ -238,6 +248,21 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
  const enIds = new Set(topo.enNodes.map((n) => n.id))
  let totalBytes = 0

+  function rowClient(r: (typeof working)[number]) {
+    const inName = resolveIfaceName(r.serverId, r.inIface).name
+    const outName = resolveIfaceName(r.serverId, r.outIface).name
+    return resolveMapClient(topo, r.serverId, inName, outName)
+  }
+
+  const payloadRows = wantDedup
+    ? dedupFlowRowsAcrossExporters(working, (a, b) => {
+      const aCli = Boolean(rowClient(a))
+      const bCli = Boolean(rowClient(b))
+      if (aCli !== bCli) return aCli ? a : b
+      return a.bytes >= b.bytes ? a : b
+    })
+    : working
+
  for (const r of working) {
    const inRes = resolveIfaceName(r.serverId, r.inIface)
    const outRes = resolveIfaceName(r.serverId, r.outIface)
@@ -323,10 +348,14 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
        }, r.bytes, "fwd")
      }
    }
+  }

+  for (const r of payloadRows) {
+    const inName = resolveIfaceName(r.serverId, r.inIface).name
+    const outName = resolveIfaceName(r.serverId, r.outIface).name
    totalBytes += r.bytes
    const peer = pickInternetPeer(r.src, r.dst, r.srcPort, r.dstPort)
-    const client = resolveClient(topo, r.serverId, inName)
+    const client = resolveMapClient(topo, r.serverId, inName, outName)
    const prevDst = dstAcc.get(peer)
    if (prevDst) {
      prevDst.bytes += r.bytes
@@ -428,10 +457,15 @@ async function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number
      const enName = nodeName(fromId)
      const viaName = nodeName(exporterId)
      for (const [clientId, c] of from.clients) {
-        const pathKey = `${clientId}|${exporterId}|${fromId}|${toId}`
+        const pathKey = `${clientId}|${fromId}|${toId}`
        const prevPath = svcPaths.get(pathKey)
        if (prevPath) {
          prevPath.bytes += c.bytes
+          if (exporterId !== fromId && prevPath.viaId === fromId) {
+            prevPath.viaId = exporterId
+            prevPath.viaName = viaName
+          }
+          if (prevPath.clientName === "—" && c.name !== "—") prevPath.clientName = c.name
        } else {
          svcPaths.set(pathKey, {
            clientId,