fix(backend): уменьшить лишние записи SQLite

Убрать WAL TRUNCATE с hot path NetFlow, не писать неизменённые UPDATE и служебные INSERT, кэшировать карты topology/analytics. Co-authored-by: Cursor <[email protected]>
fix(network-map): обрезать пунктир до сервиса и показать пути клиентов
2026-09-07 20:22:20 +07:00 · 2026-09-07 17:39:18 +07:00 · 2026-09-07 17:02:28 +07:00 · 2026-09-07 16:10:39 +07:00 · 2026-09-07 15:35:52 +07:00 · 2026-09-07 14:17:04 +07:00
39 changed files with 2953 additions and 174 deletions
@@ -14,7 +14,7 @@
    "test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts",
    "test:wireguard": "npx tsx src/services/wireguard-config.test.ts",
    "test:traffic-rate": "tsx src/services/traffic-rate.test.ts",
-    "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.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 && tsx src/services/traffic-flow-hardening.test.ts && tsx src/services/traffic-flow-purge.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-planes.test.ts && tsx src/services/traffic-flow-ip.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 && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-hardening.test.ts && tsx src/services/traffic-flow-purge.test.ts && tsx src/services/sqlite-write-opt.test.ts",
    "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts"
  },
  "dependencies": {
@@ -8,18 +8,85 @@ import { env } from "../config.js"
 import * as schema from "./schema.js"

 export const SQLITE_BUSY_TIMEOUT_MS = 5000
+/** ~16 MiB page cache (negative = KiB). */
+export const SQLITE_CACHE_SIZE_KIB = 16_000
+export const SQLITE_WAL_AUTOCHECKPOINT_PAGES = 1000
+
+export interface SqliteWriteStats {
+  insert: number
+  update: number
+  delete: number
+  walCheckpoint: number
+}
+
+let writeTrace: SqliteWriteStats | null = null
+
+function classifyWriteSql(sql: string): keyof Omit<SqliteWriteStats, "walCheckpoint"> | null {
+  const head = sql.trimStart().slice(0, 12).toUpperCase()
+  if (head.startsWith("INSERT")) return "insert"
+  if (head.startsWith("UPDATE")) return "update"
+  if (head.startsWith("DELETE")) return "delete"
+  return null
+}
+
+function installSqliteWriteTrace(handle: SqliteHandle): SqliteHandle {
+  const origPrepare = handle.prepare.bind(handle)
+  handle.prepare = ((sql: string) => {
+    const stmt = origPrepare(sql)
+    const kind = classifyWriteSql(sql)
+    if (!kind) return stmt
+    const origRun = stmt.run.bind(stmt)
+    stmt.run = ((...args: unknown[]) => {
+      if (writeTrace) writeTrace[kind] += 1
+      return origRun(...args)
+    }) as typeof stmt.run
+    return stmt
+  }) as typeof handle.prepare
+
+  const origExec = handle.exec.bind(handle)
+  handle.exec = ((sql: string) => {
+    if (writeTrace) {
+      for (const part of sql.split(";")) {
+        const kind = classifyWriteSql(part)
+        if (kind) writeTrace[kind] += 1
+      }
+    }
+    return origExec(sql)
+  }) as typeof handle.exec
+
+  const origPragma = handle.pragma.bind(handle)
+  handle.pragma = ((source: string, options?: { simple?: boolean }) => {
+    if (writeTrace && /wal_checkpoint/i.test(source)) writeTrace.walCheckpoint += 1
+    return origPragma(source, options as never)
+  }) as typeof handle.pragma
+
+  return handle
+}
+
+export function countSqliteWrites<T>(fn: () => T): { result: T; stats: SqliteWriteStats } {
+  const stats: SqliteWriteStats = { insert: 0, update: 0, delete: 0, walCheckpoint: 0 }
+  writeTrace = stats
+  try {
+    return { result: fn(), stats }
+  } finally {
+    writeTrace = null
+  }
+}

 export function applySqlitePragmas(handle: SqliteHandle): void {
  handle.pragma("journal_mode = WAL")
  handle.pragma("foreign_keys = ON")
  handle.pragma(`busy_timeout = ${SQLITE_BUSY_TIMEOUT_MS}`)
  handle.pragma("synchronous = NORMAL")
+  handle.pragma(`wal_autocheckpoint = ${SQLITE_WAL_AUTOCHECKPOINT_PAGES}`)
+  handle.pragma("temp_store = MEMORY")
+  handle.pragma(`cache_size = -${SQLITE_CACHE_SIZE_KIB}`)
 }

 function openSqlite(): SqliteHandle {
  const handle = new Database(env.DATABASE_PATH)
  applySqlitePragmas(handle)
-  return handle
+  return installSqliteWriteTrace(handle)
 }

 let sqlite = openSqlite()
@@ -146,6 +213,7 @@ CREATE TABLE IF NOT EXISTS traffic_flow_settings (
  hub_server_id INTEGER,
  retention_hours INTEGER NOT NULL DEFAULT 24,
  top_n INTEGER NOT NULL DEFAULT 200,
+  map_service_min_share_pct REAL NOT NULL DEFAULT 5,
  last_datagram_at TEXT,
  last_exporter_ip TEXT,
  last_error TEXT,
@@ -834,6 +902,13 @@ SELECT 1, 0, '10.255.254.1', 4739, 51821, '10.255.254.0/24'
 WHERE NOT EXISTS (SELECT 1 FROM traffic_flow_settings WHERE id = 1);
 `)

+{
+  const flowSettingsCols = sqlite.prepare(`PRAGMA table_info('traffic_flow_settings')`).all() as Array<{ name?: string }>
+  if (!flowSettingsCols.some((c) => c.name === "map_service_min_share_pct")) {
+    sqlite.exec(`ALTER TABLE traffic_flow_settings ADD COLUMN map_service_min_share_pct REAL NOT NULL DEFAULT 5`)
+  }
+}
+
 sqlite.exec(`
 INSERT INTO uptime_settings (id, enabled, interval_sec, retention_days)
 SELECT 1, 1, 15, 14
@@ -173,6 +173,7 @@ export const trafficFlowSettings = sqliteTable("traffic_flow_settings", {
  hubServerId:     integer("hub_server_id"),
  retentionHours:  integer("retention_hours").notNull().default(24),
  topN:            integer("top_n").notNull().default(200),
+  mapServiceMinSharePct: real("map_service_min_share_pct").notNull().default(5),
  lastDatagramAt:  text("last_datagram_at"),
  lastExporterIp:  text("last_exporter_ip"),
  lastError:       text("last_error"),
@@ -1,6 +1,7 @@
 import { desc, eq } from "drizzle-orm"
 import { db } from "../../../db/index.js"
 import { serverSnapshots, servers } from "../../../db/schema.js"
+import { invalidateFlowCatalogCache } from "../../../services/traffic-flow-topology.js"

 export type ServerRow = typeof servers.$inferSelect
 export type SnapshotRow = typeof serverSnapshots.$inferSelect
@@ -17,6 +18,7 @@ export function createServerRow(
  values: Omit<typeof servers.$inferInsert, "id">,
 ): ServerRow {
  const [inserted] = db.insert(servers).values(values).returning().all()
+  invalidateFlowCatalogCache()
  return inserted
 }

@@ -25,11 +27,13 @@ export function updateServerRowById(
  values: Partial<ServerRow>,
 ): ServerRow {
  const [updated] = db.update(servers).set(values).where(eq(servers.id, id)).returning().all()
+  invalidateFlowCatalogCache()
  return updated
 }

 export function deleteServerRowById(id: number): void {
  db.delete(servers).where(eq(servers.id, id)).run()
+  invalidateFlowCatalogCache()
 }

 export function listSnapshotsByServerId(serverId: number, limit: number): SnapshotRow[] {
@@ -1,6 +1,7 @@
-import { and, eq } from "drizzle-orm"
+import { and, count, eq } from "drizzle-orm"
 import { db } from "../../../db/index.js"
 import { appUsers, userInterfaceBindings } from "../../../db/schema.js"
+import { invalidateFlowCatalogCache } from "../../../services/traffic-flow-topology.js"

 export type AppUserRow = typeof appUsers.$inferSelect
 export type BindingRow = typeof userInterfaceBindings.$inferSelect
@@ -19,6 +20,7 @@ export function getUserRowByLogin(login: string): AppUserRow | undefined {

 export function createUserRow(values: typeof appUsers.$inferInsert): AppUserRow {
  const [inserted] = db.insert(appUsers).values(values).returning().all()
+  invalidateFlowCatalogCache()
  return inserted
 }

@@ -27,11 +29,13 @@ export function updateUserRowById(
  values: Partial<AppUserRow>,
 ): AppUserRow {
  const [updated] = db.update(appUsers).set(values).where(eq(appUsers.id, id)).returning().all()
+  invalidateFlowCatalogCache()
  return updated
 }

 export function deleteUserRowById(id: string): void {
  db.delete(appUsers).where(eq(appUsers.id, id)).run()
+  invalidateFlowCatalogCache()
 }

 export function listBindingRows(): BindingRow[] {
@@ -65,13 +69,15 @@ export function getBindingByServerIfacePeer(

 export function createBindingRow(values: typeof userInterfaceBindings.$inferInsert): BindingRow {
  const [inserted] = db.insert(userInterfaceBindings).values(values).returning().all()
+  invalidateFlowCatalogCache()
  return inserted
 }

 export function deleteBindingRowById(id: string): void {
  db.delete(userInterfaceBindings).where(eq(userInterfaceBindings.id, id)).run()
+  invalidateFlowCatalogCache()
 }

 export function countUserRows(): number {
-  return db.select().from(appUsers).all().length
+  return db.select({ n: count() }).from(appUsers).all()[0]?.n ?? 0
 }
@@ -1,4 +1,5 @@
 import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
+import { count } from "drizzle-orm"
 import { listCertificatesFromServers } from "../services/certificates-service.js"
 import { countWireGuardInterfaces } from "../services/wireguard-live.js"
 import { db } from "../db/index.js"
@@ -11,27 +12,22 @@ import {
 } from "../db/schema.js"
 import { listUsers } from "../modules/users/service/users-service.js"

+function tableCount(table: typeof servers | typeof filterRules | typeof uptimeProbes | typeof uptimeSpeedProbes | typeof recursiveRoutes): number {
+  return db.select({ n: count() }).from(table).all()[0]?.n ?? 0
+}
+
 const sidebarCountsRoutes: FastifyPluginAsyncZod = async (app) => {
  app.get("/sidebar-counts", async (_req, reply) => {
-    const [
-      serversTotal,
-      filterRulesTotal,
-      uptimeProbesTotal,
-      uptimeSpeedProbesTotal,
-      recursiveRoutesTotal,
-      certificatesTotal,
-      wireguardTotal,
-      usersTotal,
-    ] = await Promise.all([
-      Promise.resolve(db.select().from(servers).all().length),
-      Promise.resolve(db.select().from(filterRules).all().length),
-      Promise.resolve(db.select().from(uptimeProbes).all().length),
-      Promise.resolve(db.select().from(uptimeSpeedProbes).all().length),
-      Promise.resolve(db.select().from(recursiveRoutes).all().length),
+    const serversTotal = tableCount(servers)
+    const filterRulesTotal = tableCount(filterRules)
+    const uptimeProbesTotal = tableCount(uptimeProbes)
+    const uptimeSpeedProbesTotal = tableCount(uptimeSpeedProbes)
+    const recursiveRoutesTotal = tableCount(recursiveRoutes)
+    const [certificatesTotal, wireguardTotal] = await Promise.all([
      listCertificatesFromServers().then((res) => res.certificates.length),
      countWireGuardInterfaces().catch(() => 0),
-      Promise.resolve(listUsers().length),
    ])
+    const usersTotal = listUsers().length

    return reply.send({
      servers: serversTotal,
@@ -24,6 +24,7 @@ import {
  listFlowExporters,
  safeBuildLiveFlowSample,
 } from "../services/traffic-flow-analytics.js"
+import { buildFlowMapHops } from "../services/traffic-flow-map-hops.js"
 import { applyFlowOverlay } from "../services/traffic-flow-overlay.js"
 import { listTrafficFlowHostFiles } from "../services/traffic-flow-host-files.js"
 import { appendEvent } from "../modules/events/service/events-service.js"
@@ -222,6 +223,10 @@ const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => {
    return reply.send(buildFlowAnalytics(analyticsQuery(req)))
  })

+  app.get("/traffic/flow/map-hops", async (req, reply) => {
+    return reply.send(buildFlowMapHops(analyticsQuery(req)))
+  })
+
  app.get("/traffic/flow/monthly", async (req, reply) => {
    const q = req.query as { month?: string; serverId?: string }
    const now = new Date()
@@ -37,11 +37,12 @@ export function savePrevLiveMap(kind: PrevLiveKind, map: PrevLiveStringMap) {
    .limit(1)
    .all()[0]
  if (existing) {
+    if (existing.payloadJson === payloadJson) return
    db.update(alertEnginePrevLive)
      .set({ payloadJson, updatedAt: new Date().toISOString() })
      .where(eq(alertEnginePrevLive.kind, kind))
      .run()
-  } else {
-    db.insert(alertEnginePrevLive).values({ kind, payloadJson, updatedAt: new Date().toISOString() }).run()
+    return
  }
+  db.insert(alertEnginePrevLive).values({ kind, payloadJson, updatedAt: new Date().toISOString() }).run()
 }
@@ -1,5 +1,3 @@
-import { db, sqliteDatabase } from "../db/index.js"
-import { alertBgpPeerSamples, alertGreTunnelSamples } from "../db/schema.js"
 import { bgpPeerAlertKey, fetchBgpSessionsForAlerts } from "./bgp-peers-live.js"
 import { fetchGreTunnelLiveRows } from "./gre-tunnels-live.js"
 import type { GreBgpSnapshotRunSnapshot } from "../types/scheduler-run-snapshot.js"
@@ -12,8 +10,8 @@ export function getGreBgpSnapshotCollectorState(): { running: boolean } {
 }

 /**
- * Один опрос GRE + BGP по включённым серверам и запись строк в SQLite для `buildSignalSnapshot`.
- * Движок оповещений больше не дублирует эти REST-запросы.
+ * Один опрос GRE + BGP по включённым серверам.
+ * Снимок для алертов живёт в `scheduler_runs.result_json` (`greTunnels` / `bgpPeers`).
 */
 export async function collectGreBgpSnapshotOnce(): Promise<GreBgpSnapshotRunSnapshot> {
  const sampledAt = new Date().toISOString()
@@ -62,32 +60,8 @@ export async function collectGreBgpSnapshotOnce(): Promise<GreBgpSnapshotRunSnap
    const bgpRows = bgpSettled.status === "fulfilled" ? bgpSettled.value : []
    snapshot.greTunnels = greRows.map((r) => ({ targetLabel: r.targetLabel, status: r.status }))
    snapshot.bgpPeers = bgpRows.map((s) => ({ key: bgpPeerAlertKey(s), state: s.state }))
-
-    db.transaction((tx) => {
-      for (const r of greRows) {
-        tx.insert(alertGreTunnelSamples).values({
-          sampledAt,
-          targetLabel: r.targetLabel,
-          status: r.status,
-        }).run()
-        snapshot.greWritten += 1
-      }
-      for (const s of bgpRows) {
-        tx.insert(alertBgpPeerSamples).values({
-          sampledAt,
-          peerKey: bgpPeerAlertKey(s),
-          state: s.state,
-        }).run()
-        snapshot.bgpWritten += 1
-      }
-    })
-
-    sqliteDatabase
-      .prepare(`DELETE FROM alert_gre_tunnel_samples WHERE sampled_at < datetime('now', '-30 days')`)
-      .run()
-    sqliteDatabase
-      .prepare(`DELETE FROM alert_bgp_peer_samples WHERE sampled_at < datetime('now', '-30 days')`)
-      .run()
+    snapshot.greWritten = greRows.length
+    snapshot.bgpWritten = bgpRows.length

    if (errors.length) snapshot.errors = errors
  } catch (e) {
@@ -5,6 +5,7 @@ import type { SnapshotInsert } from "../db/schema.js"
 import type { SnapshotRead } from "../types/server.js"
 import { MikrotikClient } from "./mikrotik.js"
 import { parseRosCpuLoadPercent, parseRosDataSizeBytes } from "./ros-metric-parse.js"
+import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js"

 // ── pollServer ─────────────────────────────────────────────────────────────────

@@ -64,11 +65,13 @@ export async function pollServer(serverId: number): Promise<SnapshotRead> {
      rawIpAddresses: JSON.stringify(addresses),
    } satisfies Partial<SnapshotInsert>)

-    // Keep server.name in sync with RouterOS identity
-    db.update(servers)
-      .set({ name: identity.name, updatedAt: now })
-      .where(eq(servers.id, serverId))
-      .run()
+    if ((identity.name || "") !== (server.name || "")) {
+      db.update(servers)
+        .set({ name: identity.name, updatedAt: now })
+        .where(eq(servers.id, serverId))
+        .run()
+      invalidateFlowCatalogCache()
+    }

  } catch (err) {
    // Log but don't throw — we still persist the offline snapshot
@@ -7,7 +7,7 @@
 * 3. `uptime_speed` — ниже (BW-test, тяжёлый); локи узлов — `withBtestNodeLocks` в speed-сервисе.
 *
 * **Оповещения (`alert_engine`):** читают SQLite после джоб сбора (см. `buildSignalSnapshot`), в т.ч.
- * `gre_bgp` → `alert_gre_tunnel_samples` / `alert_bgp_peer_samples`. После успешного завершения джоб
+ * `gre_bgp` → snapshot в `scheduler_runs.result_json` (`greTunnels` / `bgpPeers`). После успешного завершения джоб
 * `traffic`, `uptime_*`, `servers_rest_ping`, `gre_bgp` планируется **дополнительный** прогон движка
 * (debounce), см. [`alert-collector-hooks.ts`](./alert-collector-hooks.ts); любой другой писатель сэмплов
 * для снимка оповещений тоже должен вызывать `scheduleAlertEngineAfterDataCollectors()` после коммита.
@@ -17,7 +17,7 @@
 */
 import { desc, eq, lt } from "drizzle-orm"
 import { db } from "../db/index.js"
-import { schedulerRuns } from "../db/schema.js"
+import { events, schedulerRuns } from "../db/schema.js"
 import type { SchedulerRunSnapshot } from "../types/scheduler-run-snapshot.js"
 import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
 import {
@@ -71,6 +71,20 @@ const timers = new Map<string, ReturnType<typeof setInterval>>()

 const RUN_LOG_RETENTION_MS = 30 * 24 * 60 * 60 * 1000

+const QUIET_SCHEDULER_OK_JOBS = new Set<SchedulerJobKey>([
+  "traffic",
+  "servers_rest_ping",
+  "uptime_resources",
+  "uptime_ping",
+  "uptime_speed",
+  "gre_bgp",
+  "alert_engine",
+])
+
+export function shouldAppendSchedulerOkEvent(jobKey: SchedulerJobKey): boolean {
+  return !QUIET_SCHEDULER_OK_JOBS.has(jobKey)
+}
+
 function newRunId(): string {
  return `sch-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`
 }
@@ -84,18 +98,21 @@ function appendSchedulerRun(row: {
  durationMs: number
  result?: SchedulerRunSnapshot | null
 }) {
-  db.insert(schedulerRuns).values({
-    id: newRunId(),
-    jobKey: row.jobKey,
-    startedAt: row.startedAt,
-    finishedAt: row.finishedAt,
-    status: row.status,
-    error: row.error,
-    durationMs: row.durationMs,
-    resultJson: row.result ? JSON.stringify(row.result) : null,
-  }).run()
  const cutoff = new Date(Date.now() - RUN_LOG_RETENTION_MS).toISOString()
-  db.delete(schedulerRuns).where(lt(schedulerRuns.finishedAt, cutoff)).run()
+  db.transaction((tx) => {
+    tx.insert(schedulerRuns).values({
+      id: newRunId(),
+      jobKey: row.jobKey,
+      startedAt: row.startedAt,
+      finishedAt: row.finishedAt,
+      status: row.status,
+      error: row.error,
+      durationMs: row.durationMs,
+      resultJson: row.result ? JSON.stringify(row.result) : null,
+    }).run()
+    tx.delete(schedulerRuns).where(lt(schedulerRuns.finishedAt, cutoff)).run()
+    tx.delete(events).where(lt(events.createdAt, cutoff)).run()
+  })
 }

 async function runSchedulerJobBody(jobKey: SchedulerJobKey): Promise<void> {
@@ -159,19 +176,21 @@ async function runSchedulerJobBody(jobKey: SchedulerJobKey): Promise<void> {
      durationMs: Date.now() - startedAt,
      result: snapshot ?? null,
    })
-    appendEvent({
-      level: "info",
-      eventType: "scheduler.job.ok",
-      sourceModule: "scheduler",
-      title: "Задача планировщика завершена",
-      message: `${jobKey}: выполнено за ${Date.now() - startedAt} мс`,
-      entityType: "job",
-      entityId: jobKey,
-      payload: {
-        startedAt: startedIso,
-        finishedAt: finishedIso,
-      },
-    })
+    if (shouldAppendSchedulerOkEvent(jobKey)) {
+      appendEvent({
+        level: "info",
+        eventType: "scheduler.job.ok",
+        sourceModule: "scheduler",
+        title: "Задача планировщика завершена",
+        message: `${jobKey}: выполнено за ${Date.now() - startedAt} мс`,
+        entityType: "job",
+        entityId: jobKey,
+        payload: {
+          startedAt: startedIso,
+          finishedAt: finishedIso,
+        },
+      })
+    }
    if (
      jobKey === "traffic" ||
      jobKey === "servers_rest_ping" ||
@@ -0,0 +1,144 @@
+import assert from "node:assert/strict"
+import { randomUUID } from "node:crypto"
+import { countSqliteWrites, sqliteDatabase } from "../db/index.js"
+import { events } from "../db/schema.js"
+import { db } from "../db/index.js"
+import {
+  attachEngineSqlite,
+  bumpPacketMeta,
+  configureEngine,
+  flushPending,
+  ingestParsedFlowsForServerForTests,
+  resetEngineForTests,
+  setEngineError,
+} from "./traffic-flow-engine.js"
+import { collectGreBgpSnapshotOnce } from "./gre-bgp-snapshot-collector.js"
+import { shouldAppendSchedulerOkEvent, executeSchedulerJob } from "./scheduler.js"
+import { savePrevLiveMap, loadPrevLiveMap } from "./alert-engine/prev-live-store.js"
+import {
+  invalidateFlowCatalogCache,
+  loadFlowTopology,
+  seedFlowTopologyForTests,
+} from "./traffic-flow-topology.js"
+
+resetEngineForTests()
+attachEngineSqlite(sqliteDatabase)
+seedFlowTopologyForTests(null)
+invalidateFlowCatalogCache()
+
+const idle1 = countSqliteWrites(() => {
+  flushPending()
+})
+assert.equal(idle1.stats.walCheckpoint, 0)
+assert.ok(idle1.stats.update >= 1, "first idle flush persists listener stats")
+
+const idle2 = countSqliteWrites(() => {
+  flushPending()
+})
+assert.equal(idle2.stats.update, 0, "unchanged listener stats skip UPDATE")
+assert.equal(idle2.stats.walCheckpoint, 0)
+
+bumpPacketMeta("203.0.113.9")
+const changed = countSqliteWrites(() => {
+  flushPending()
+})
+assert.equal(changed.stats.update, 1, "changed packets persist once")
+assert.equal(changed.stats.walCheckpoint, 0)
+
+setEngineError("boom")
+const errWrite = countSqliteWrites(() => {
+  flushPending()
+})
+assert.equal(errWrite.stats.update, 1)
+setEngineError("")
+flushPending()
+
+resetEngineForTests()
+configureEngine({ topN: 20 })
+ingestParsedFlowsForServerForTests(9, [{
+  src: "10.1.1.1",
+  dst: "8.8.8.8",
+  proto: 6,
+  srcPort: 40000,
+  dstPort: 443,
+  bytes: 100,
+  packets: 1,
+  inIface: "2",
+  outIface: "",
+}])
+const withData = countSqliteWrites(() => {
+  flushPending()
+})
+assert.equal(withData.stats.walCheckpoint, 0, "flush with data must not TRUNCATE WAL")
+assert.ok(withData.stats.insert >= 1, "flow upsert writes")
+
+sqliteDatabase.prepare(`DELETE FROM flow_buckets WHERE server_id = 9`).run()
+sqliteDatabase.prepare(`DELETE FROM flow_minute_stats WHERE server_id = 9`).run()
+sqliteDatabase.prepare(`DELETE FROM flow_minute_dims WHERE server_id = 9`).run()
+sqliteDatabase.prepare(`DELETE FROM flow_daily_dims WHERE server_id = 9`).run()
+
+const plan = sqliteDatabase.prepare(`
+  EXPLAIN QUERY PLAN
+  SELECT id FROM flow_buckets WHERE bucket_at >= ? ORDER BY bytes DESC LIMIT 100
+`).all("2000-01-01T00:00:00.000Z") as Array<{ detail?: string }>
+const planText = plan.map((p) => String(p.detail ?? "")).join(" | ")
+assert.ok(planText.length > 0, "EXPLAIN QUERY PLAN returned rows")
+
+invalidateFlowCatalogCache()
+seedFlowTopologyForTests(null)
+const topoA = loadFlowTopology()
+const topoB = loadFlowTopology()
+assert.equal(topoA, topoB, "topology cache returns same object")
+invalidateFlowCatalogCache()
+const topoC = loadFlowTopology()
+assert.notEqual(topoA, topoC, "invalidate rebuilds topology")
+
+assert.equal(shouldAppendSchedulerOkEvent("traffic"), false)
+assert.equal(shouldAppendSchedulerOkEvent("alert_engine"), false)
+assert.equal(shouldAppendSchedulerOkEvent("gre_bgp"), false)
+assert.equal(shouldAppendSchedulerOkEvent("backups"), true)
+assert.equal(shouldAppendSchedulerOkEvent("certificates_renew"), true)
+assert.equal(shouldAppendSchedulerOkEvent("internet_path"), true)
+
+savePrevLiveMap("gre", { a: "up" })
+const prevSame = countSqliteWrites(() => {
+  savePrevLiveMap("gre", { a: "up" })
+})
+assert.equal(prevSame.stats.update, 0)
+assert.equal(prevSame.stats.insert, 0)
+savePrevLiveMap("gre", { a: "down" })
+assert.equal(loadPrevLiveMap("gre").a, "down")
+savePrevLiveMap("gre", { a: "up" })
+
+const greBefore = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_gre_tunnel_samples`).get() as { n: number }
+const bgpBefore = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_bgp_peer_samples`).get() as { n: number }
+await collectGreBgpSnapshotOnce()
+const greAfter = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_gre_tunnel_samples`).get() as { n: number }
+const bgpAfter = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_bgp_peer_samples`).get() as { n: number }
+assert.equal(greAfter.n, greBefore.n)
+assert.equal(bgpAfter.n, bgpBefore.n)
+
+const oldId = `evt-old-${randomUUID()}`
+db.insert(events).values({
+  id: oldId,
+  createdAt: new Date(Date.now() - 40 * 24 * 60 * 60 * 1000).toISOString(),
+  level: "info",
+  eventType: "test.retention",
+  sourceModule: "system",
+  title: "old",
+  message: "old",
+}).run()
+const eventsBefore = sqliteDatabase.prepare(
+  `SELECT COUNT(*) AS n FROM events WHERE event_type = 'scheduler.job.ok' AND entity_id = 'alert_engine'`,
+).get() as { n: number }
+await executeSchedulerJob("alert_engine")
+const oldGone = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM events WHERE id = ?`).get(oldId) as { n: number }
+assert.equal(oldGone.n, 0, "events older than 30 days are purged")
+const eventsAfter = sqliteDatabase.prepare(
+  `SELECT COUNT(*) AS n FROM events WHERE event_type = 'scheduler.job.ok' AND entity_id = 'alert_engine'`,
+).get() as { n: number }
+assert.equal(eventsAfter.n, eventsBefore.n, "quiet jobs do not append scheduler.job.ok")
+
+resetEngineForTests()
+console.log("sqlite-write-opt.test.ts: ok")
+console.log("EXPLAIN listStoredFlowRows:", planText)
@@ -384,6 +384,51 @@ try {
  }
 }

+{
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  disableRipeEnqueueForTests()
+  rememberServerIfaces(7, [{ ".id": "*2", name: "ether1" }])
+  ingestParsedFlowsForServerForTests(7, [
+    {
+      src: "173.194.151.65",
+      dst: "10.200.100.53",
+      proto: 6,
+      srcPort: 443,
+      dstPort: 57182,
+      bytes: 12_000,
+      packets: 10,
+      inIface: "2",
+      outIface: "2",
+    },
+    {
+      src: "104.18.35.51",
+      dst: "10.200.100.53",
+      proto: 6,
+      srcPort: 443,
+      dstPort: 53880,
+      bytes: 3_000,
+      packets: 4,
+      inIface: "2",
+      outIface: "2",
+    },
+  ])
+  try {
+    const rev = buildFlowAnalytics({ minutes: 5, serverId: 7 })
+    const google = rev.conversationsList.find((r) => r.src === "173.194.151.65")
+    const cf = rev.conversationsList.find((r) => r.src === "104.18.35.51")
+    assert.equal(google?.service, "Google")
+    assert.equal(google?.category, "Веб")
+    assert.equal(cf?.service, "Cloudflare")
+    assert.equal(cf?.category, "CDN")
+  } finally {
+    resetFlowRingsForTests()
+    resetIfaceCacheForTests()
+    resetRipeCacheForTests()
+  }
+}
+
 {
  resetFlowRingsForTests()
  resetIfaceCacheForTests()
@@ -1,6 +1,6 @@
 import { eq } from "drizzle-orm"
 import { db, sqliteDatabase } from "../db/index.js"
-import { appUsers, servers, userInterfaceBindings } from "../db/schema.js"
+import { appUsers, userInterfaceBindings } from "../db/schema.js"
 import type {
  FlowAnalyticsDto,
  FlowBreakdownRow,
@@ -21,7 +21,7 @@ import {
  listFlowRowsForWindow,
  type PendingFlowRow,
 } from "./traffic-flow-ingest.js"
-import { MAX_PENDING, RING_OVERLAY } from "./traffic-flow-engine.js"
+import { flowDataEpoch, MAX_PENDING, RING_OVERLAY } from "./traffic-flow-engine.js"
 import { resolveIfaceName } from "./traffic-flow-ifaces.js"
 import { getTrafficFlowSettingsRow, listHostPeers } from "./traffic-flow-settings.js"
 import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
@@ -30,8 +30,10 @@ import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js"
 import { classifyFlowDst, refreshFlowCatalogInBackground } from "./traffic-flow-classify.js"
 import { isIsoCountry } from "./traffic-flow-brands.js"
 import { classifyFlowPlane, flowBps, shouldKeepPlane } from "./traffic-flow-planes.js"
+import { pickInternetPeer } from "./traffic-flow-ip.js"
 import {
  enGreIfaceNames,
+  getServerCatalog,
  latestWireBps,
  loadFlowTopology,
  resolveClient,
@@ -141,14 +143,46 @@ function topLabel(map: Map<string, { bytes: number; packets: number; label?: str
 }

 export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
+  const key = analyticsQueryKey(q)
+  const now = Date.now()
+  if (analyticsCache && analyticsCache.key === key && now - analyticsCache.at < ANALYTICS_CACHE_TTL_MS) {
+    return analyticsCache.dto
+  }
+  const dto = buildFlowAnalyticsUncached(q)
+  analyticsCache = { key, at: now, dto }
+  return dto
+}
+
+export function resetFlowAnalyticsCacheForTests(): void {
+  analyticsCache = null
+}
+
+function analyticsQueryKey(q: FlowAnalyticsQuery): string {
+  return JSON.stringify({
+    epoch: flowDataEpoch(),
+    minutes: q.minutes,
+    serverId: q.serverId ?? null,
+    userId: q.userId ?? null,
+    iface: q.iface ?? null,
+    dedup: q.dedup !== false,
+    excludeMesh: q.excludeMesh !== false,
+    excludeOverlay: q.excludeOverlay !== false,
+    skipHeavy: Boolean(q.skipHeavy),
+  })
+}
+
+const ANALYTICS_CACHE_TTL_MS = 2000
+let analyticsCache: { key: string; at: number; dto: FlowAnalyticsDto } | null = null
+
+function buildFlowAnalyticsUncached(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 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 catalog = getServerCatalog()
+  const nameById = new Map([...catalog.byId].map(([id, s]) => [id, s.name]))
+  const countryById = new Map([...catalog.byId].map(([id, s]) => [id, s.country]))
  const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : ""
  const wantDedup = q.dedup !== false && !ifaceFilter
  const excludeMesh = q.excludeMesh !== false
@@ -171,6 +205,7 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
  const pathAcc = new Map<string, FlowPathRow>()
  const srcs = new Set<string>()
  const dsts = new Set<string>()
+  const peers = new Set<string>()
  const matched: PendingFlowRow[] = []
  const skipHeavy = Boolean(q.skipHeavy)
  let bytesPayload = 0
@@ -217,9 +252,11 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
    totalPackets += r.packets
    srcs.add(r.src)
    dsts.add(r.dst)
+    const peer = pickInternetPeer(r.src, r.dst, r.srcPort, r.dstPort)
+    peers.add(peer)
    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)
+    const ripe = lookupRipeCached(peer)
+    const classified = classifyFlowDst(peer, 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)
@@ -345,7 +382,7 @@ export function buildFlowAnalytics(q: FlowAnalyticsQuery): FlowAnalyticsDto {
    }
  }

-  enqueueRipeMisses(dsts)
+  enqueueRipeMisses(peers)

  const conversationsList = [...conv.values()]
    .map((t) => {
@@ -474,19 +511,19 @@ export function listFlowExporters(minutes: number): FlowExportersDto {
  const { bytes, sessions } = summarizeByServer(rows)
  const ids = new Set<number>([...bytes.keys()])
  for (const p of listHostPeers()) ids.add(p.serverId)
-  const serverRows = db.select().from(servers).all()
+  const catalog = getServerCatalog()
  const emptySeries = Array(60).fill(0) as number[]
-  const exporters = serverRows
+  const exporters = catalog.list
    .filter((s) => ids.has(s.id))
    .map((s) => {
      const ring = getRingMbps(s.id, "__all__")
      const total = bytes.get(s.id) ?? 0
      return {
        id: String(s.id),
-        name: s.name || s.host,
+        name: s.name,
        subtitle: s.host,
-        site: s.site || "—",
-        country: s.country || "UN",
+        site: s.site,
+        country: s.country,
        status: snapshotStatus(s.id),
        rxNow: ring.rxNow || (total * 8) / Math.max(60, minutes * 60) / 1_000_000,
        txNow: ring.txNow,
@@ -4,6 +4,8 @@ import {
  countryFromHolder,
  lookupBrand,
  OTHER_SERVICE,
+  isNamedInternetService,
+  mapServiceNodeId,
  resolveRipeCountry,
 } from "./traffic-flow-brands.js"

@@ -21,9 +23,20 @@ assert.equal(brandByAsn(15169)?.category, "Веб")
 assert.equal(lookupBrand("208.65.153.1", 0)?.service, "YouTube")
 assert.equal(brandByAsn(32590)?.service, "Steam")
 assert.equal(brandByAsn(32590)?.category, "Игры")
+assert.equal(brandByAsn(16509)?.service, "AWS")
+assert.equal(brandByAsn(57976)?.service, "Blizzard")
 assert.equal(brandByAsn(401115)?.service, "ChatGPT")
 assert.equal(lookupBrand("1.1.1.1", 13335)?.service, "Cloudflare")
+assert.equal(lookupBrand("104.18.35.51", 0)?.service, "Cloudflare")
+assert.equal(lookupBrand("173.194.151.65", 0)?.service, "Google")
+assert.equal(lookupBrand("8.8.8.8", 0)?.service, "Google")
 assert.equal(lookupBrand("203.0.113.9", 64500), null)
 assert.equal(OTHER_SERVICE, "Прочее")
+assert.equal(isNamedInternetService("Google", "Веб"), true)
+assert.equal(isNamedInternetService("Прочее", "Прочее"), false)
+assert.equal(isNamedInternetService("GRE", "Туннель"), false)
+assert.equal(isNamedInternetService("DNS", "DNS"), false)
+assert.equal(mapServiceNodeId("AWS"), "svc:aws")
+assert.equal(mapServiceNodeId("Cloudflare"), "svc:cloudflare")

 console.log("traffic-flow-brands.test.ts: ok")
@@ -12,11 +12,12 @@ const ASN_BRANDS = new Map<number, BrandHit>([
  [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" }],
+  [16509, { service: "AWS", category: "CDN" }],
+  [14618, { service: "AWS", category: "CDN" }],
  [8075, { service: "Microsoft", category: "CDN" }],
  [13238, { service: "Yandex", category: "CDN" }],
  [32590, { service: "Steam", category: "Игры" }],
+  [57976, { service: "Blizzard", category: "Игры" }],
  [2906, { service: "Netflix", category: "Видео / стриминг" }],
  [40027, { service: "Netflix", category: "Видео / стриминг" }],
  [15169, { service: "Google", category: "Веб" }],
@@ -41,6 +42,7 @@ const ASN_HQ_COUNTRY = new Map<number, string>([
  [8075, "US"],
  [15169, "US"],
  [32590, "US"],
+  [57976, "US"],
  [2906, "US"],
  [40027, "US"],
  [36040, "US"],
@@ -54,13 +56,23 @@ const ASN_HQ_COUNTRY = new Map<number, string>([
  [211157, "NL"],
 ])

+const GOOGLE: BrandHit = { service: "Google", category: "Веб" }
+const CLOUDFLARE: BrandHit = { service: "Cloudflare", category: "CDN" }
+const YOUTUBE: BrandHit = { service: "YouTube", category: "Видео / стриминг" }
+
 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" } },
-  { cidr: "208.65.152.0/22", prefixLen: 22, hit: { service: "YouTube", category: "Видео / стриминг" } },
-  { cidr: "208.117.224.0/19", prefixLen: 19, hit: { service: "YouTube", category: "Видео / стриминг" } },
+  { cidr: "104.16.0.0/13", prefixLen: 13, hit: CLOUDFLARE },
+  { cidr: "104.24.0.0/14", prefixLen: 14, hit: CLOUDFLARE },
+  { cidr: "172.64.0.0/13", prefixLen: 13, hit: CLOUDFLARE },
+  { cidr: "162.158.0.0/15", prefixLen: 15, hit: CLOUDFLARE },
+  { cidr: "8.8.8.0/24", prefixLen: 24, hit: GOOGLE },
+  { cidr: "8.8.4.0/24", prefixLen: 24, hit: GOOGLE },
+  { cidr: "173.194.0.0/16", prefixLen: 16, hit: GOOGLE },
+  { cidr: "172.217.0.0/16", prefixLen: 16, hit: GOOGLE },
+  { cidr: "74.125.0.0/16", prefixLen: 16, hit: GOOGLE },
+  { cidr: "142.250.0.0/15", prefixLen: 15, hit: GOOGLE },
+  { cidr: "208.65.152.0/22", prefixLen: 22, hit: YOUTUBE },
+  { cidr: "208.117.224.0/19", prefixLen: 19, hit: YOUTUBE },
 ].sort((a, b) => b.prefixLen - a.prefixLen)

 const NON_ISO = new Set(["EU", "AP", "ZZ", "XX", "A1", "A2", "O1"])
@@ -105,3 +117,32 @@ export function brandByCidr(ip: string): BrandHit | null {
 export function lookupBrand(ip: string, asn: number): BrandHit | null {
  return brandByCidr(ip) || brandByAsn(asn)
 }
+
+const SKIP_MAP_SERVICES = new Set([
+  OTHER_SERVICE,
+  "GRE",
+  "ESP",
+  "WireGuard",
+  "DNS",
+  "SSH",
+  "BGP",
+])
+
+const SKIP_MAP_CATEGORIES = new Set(["Туннель", "DNS", "SSH", "BGP"])
+
+/** Именованный интернет-сервис для карты (не туннель и не «Прочее»). */
+export function isNamedInternetService(service: string, category: string): boolean {
+  const s = service.trim()
+  const c = category.trim()
+  if (!s || SKIP_MAP_SERVICES.has(s) || SKIP_MAP_CATEGORIES.has(c)) return false
+  return true
+}
+
+export function mapServiceNodeId(label: string): string {
+  const slug = label
+    .trim()
+    .toLowerCase()
+    .replace(/[^a-z0-9]+/g, "-")
+    .replace(/^-+|-+$/g, "")
+  return `svc:${slug || "unknown"}`
+}
@@ -36,6 +36,10 @@ const google = classifyFlowDst("173.194.160.163", 6, 443, 1, {
 assert.equal(google.service, "Google")
 assert.equal(google.category, "Веб")

+const googleCidr = classifyFlowDst("173.194.151.65", 6, 57182, 443, null)
+assert.equal(googleCidr.service, "Google")
+assert.equal(googleCidr.category, "Веб")
+
 const youtube = classifyFlowDst("173.194.160.163", 6, 443, 1, {
  prefix: "173.194.0.0/16",
  asn: 15169,
@@ -4,9 +4,11 @@ import { classifyFlowPlaneLite } from "./traffic-flow-planes.js"
 import { pickServerIdForExporter, type OverlayPeerRef } from "./traffic-flow-map-exporter.js"
 import { applicationName } from "./traffic-flow-apps.js"
 import { classifyFlowDst } from "./traffic-flow-classify.js"
-import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js"
+import { enqueueRipeMisses, lookupRipeCached, pruneRipeSqlite } from "./traffic-flow-ripe.js"
+import { invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js"
 import { isIsoCountry } from "./traffic-flow-brands.js"
 import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js"
+import { pickInternetPeer } from "./traffic-flow-ip.js"

 type SqliteHandle = InstanceType<typeof Database>

@@ -93,10 +95,27 @@ let dropped = 0
 let rowsStored = 0
 let lastFlushUsedTransaction = false
 let lastPruneAt = 0
+let lastPassiveCheckpointAt = Date.now()
+let dataEpoch = 0
+let lastPersistedStats: {
+  packetsReceived: number
+  lastDatagramAt: string | null
+  lastExporterIp: string | null
+  lastError: string
+} | null = null
 let exporterCtx: ExporterResolveCtx | null = null

 const PRUNE_MS = 5 * 60_000
 const LIVE_WINDOW_MS = 15 * 60_000
+const PASSIVE_CHECKPOINT_MS = 60_000
+
+function bumpDataEpoch(): void {
+  dataEpoch += 1
+}
+
+export function flowDataEpoch(): number {
+  return dataEpoch
+}

 function nowIso(): string {
  return new Date().toISOString()
@@ -233,15 +252,17 @@ export function getEngineStats(): EngineStats {
 }

 export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): void {
+  if (flows.length) bumpDataEpoch()
  const bucketAt = minuteBucketIso()
  const ripeMisses: string[] = []
  for (const raw of flows) {
    const flow = normalizeParsedFlow(raw)
    addToTick(serverId, flow, flow.bytes)
    bumpRollup(serverId, bucketAt, flow, flow.bytes, flow.packets)
-    const ripe = lookupRipeCached(flow.dst)
-    if (flow.dst && !ripe) ripeMisses.push(flow.dst)
-    const classified = classifyFlowDst(flow.dst, flow.proto, flow.dstPort, flow.srcPort, ripe)
+    const peer = pickInternetPeer(flow.src, flow.dst, flow.srcPort, flow.dstPort)
+    const ripe = lookupRipeCached(peer)
+    if (peer && !ripe) ripeMisses.push(peer)
+    const classified = classifyFlowDst(peer, flow.proto, flow.dstPort, flow.srcPort, ripe)
    const app = applicationName(flow.proto, flow.dstPort, flow.srcPort)
    const country = ripe?.ok && isIsoCountry(ripe.country)
      ? ripe.country
@@ -420,7 +441,16 @@ export function applyRingSnapshot(rows: Array<{ key: string; inBps: number[]; ou
  }
 }

-function persistListenerStats(handle: SqliteHandle): void {
+function persistListenerStats(handle: SqliteHandle): boolean {
+  if (
+    lastPersistedStats
+    && lastPersistedStats.packetsReceived === packetsReceived
+    && lastPersistedStats.lastDatagramAt === lastDatagramAt
+    && lastPersistedStats.lastExporterIp === lastExporterIp
+    && lastPersistedStats.lastError === lastError
+  ) {
+    return false
+  }
  handle.prepare(`
    UPDATE traffic_flow_settings
    SET packets_received = @packetsReceived,
@@ -436,6 +466,25 @@ function persistListenerStats(handle: SqliteHandle): void {
    lastError,
    updatedAt: nowIso(),
  })
+  lastPersistedStats = {
+    packetsReceived,
+    lastDatagramAt,
+    lastExporterIp,
+    lastError,
+  }
+  invalidateTrafficFlowSettingsCache()
+  return true
+}
+
+function maybePassiveCheckpoint(handle: SqliteHandle): void {
+  const now = Date.now()
+  if (now - lastPassiveCheckpointAt < PASSIVE_CHECKPOINT_MS) return
+  lastPassiveCheckpointAt = now
+  try {
+    handle.pragma("wal_checkpoint(PASSIVE)")
+  } catch {
+    /* ignore */
+  }
 }

 function upsertMinuteAndDaily(handle: SqliteHandle): void {
@@ -584,6 +633,7 @@ function pruneStored(handle: SqliteHandle): void {
      }
    }
  }
+  pruneRipeSqlite(now)
 }

 function topNPending(rows: PendingFlowRow[]): PendingFlowRow[] {
@@ -614,6 +664,7 @@ export function flushPending(): void {
  persistListenerStats(handle)
  if (pending.size === 0 && minuteRollup.size === 0 && minuteDims.size === 0) {
    pruneStored(handle)
+    maybePassiveCheckpoint(handle)
    lastFlushUsedTransaction = false
    return
  }
@@ -663,6 +714,7 @@ export function flushPending(): void {
    tx(rows)
    lastFlushUsedTransaction = true
    rowsStored += rows.length
+    bumpDataEpoch()
  } catch {
    for (const r of rows) {
      try {
@@ -695,11 +747,7 @@ export function flushPending(): void {
    /* rollup best-effort */
  }
  pruneStored(handle)
-  try {
-    handle.pragma("wal_checkpoint(TRUNCATE)")
-  } catch {
-    /* ignore */
-  }
+  maybePassiveCheckpoint(handle)
 }

 export function lastFlushUsedTransactionForTests(): boolean {
@@ -734,6 +782,9 @@ export function resetEngineForTests(): void {
  rowsStored = 0
  lastFlushUsedTransaction = false
  lastPruneAt = 0
+  lastPassiveCheckpointAt = Date.now()
+  lastPersistedStats = null
+  bumpDataEpoch()
  pendingCap = MAX_PENDING
 }

@@ -1,5 +1,10 @@
 import assert from "node:assert/strict"
-import { SQLITE_BUSY_TIMEOUT_MS, sqliteDatabase } from "../db/index.js"
+import {
+  SQLITE_BUSY_TIMEOUT_MS,
+  SQLITE_CACHE_SIZE_KIB,
+  SQLITE_WAL_AUTOCHECKPOINT_PAGES,
+  sqliteDatabase,
+} from "../db/index.js"
 import {
  MAX_FLOW_LIVE_SUBSCRIBERS,
  resetFlowLiveSlotsForTests,
@@ -11,6 +16,16 @@ const busy = sqliteDatabase.pragma("busy_timeout") as Array<{ busy_timeout: numb
 const busyValue = Array.isArray(busy) ? Number(Object.values(busy[0] ?? {})[0]) : Number(busy)
 assert.equal(busyValue, SQLITE_BUSY_TIMEOUT_MS)

+function pragmaNum(name: string): number {
+  const rows = sqliteDatabase.pragma(name) as Array<Record<string, number>>
+  const row = Array.isArray(rows) ? rows[0] : rows
+  return Number(Object.values(row ?? {})[0])
+}
+
+assert.equal(pragmaNum("wal_autocheckpoint"), SQLITE_WAL_AUTOCHECKPOINT_PAGES)
+assert.equal(pragmaNum("cache_size"), -SQLITE_CACHE_SIZE_KIB)
+assert.equal(pragmaNum("temp_store"), 2)
+
 resetFlowLiveSlotsForTests()
 for (let i = 0; i < MAX_FLOW_LIVE_SUBSCRIBERS; i++) {
  assert.equal(tryAcquireFlowLiveSlot(), true)
@@ -33,6 +33,7 @@ import {
 } from "./traffic-flow-settings.js"
 import { applicationName } from "./traffic-flow-apps.js"
 import { resolveIfaceName } from "./traffic-flow-ifaces.js"
+import { getServerCatalog } from "./traffic-flow-topology.js"

 export type { PendingFlowRow }

@@ -332,8 +333,8 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto {
  const settings = getTrafficFlowSettingsRow()
  const runtime = getFlowRuntimeCounters()
  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 catalog = getServerCatalog()
+  const nameById = new Map([...catalog.byId].map(([id, s]) => [id, s.name]))
  const agg = new Map<string, FlowTalkerDto & { rawBytes: number }>()
  const protoBytes = new Map<number, number>()
  const srcs = new Set<string>()
@@ -0,0 +1,25 @@
+import assert from "node:assert/strict"
+import { isNonPublicIp, pickInternetPeer } from "./traffic-flow-ip.js"
+
+assert.equal(isNonPublicIp("10.200.100.53"), true)
+assert.equal(isNonPublicIp("173.194.151.65"), false)
+
+assert.equal(
+  pickInternetPeer("173.194.151.65", "10.200.100.53", 443, 57182),
+  "173.194.151.65",
+  "reverse IPFIX: Google:443 → RFC1918",
+)
+assert.equal(
+  pickInternetPeer("10.200.100.53", "104.18.35.51", 53880, 443),
+  "104.18.35.51",
+  "client → Cloudflare:443",
+)
+assert.equal(pickInternetPeer("10.100.1.17", "8.8.8.8", 51234, 443), "8.8.8.8")
+assert.equal(
+  pickInternetPeer("1.1.1.1", "8.8.8.8", 443, 51234),
+  "1.1.1.1",
+  "оба публичные — сторона с well-known портом",
+)
+assert.equal(pickInternetPeer("10.1.1.1", "10.2.2.2", 443, 80), "10.2.2.2")
+
+console.log("traffic-flow-ip.test.ts: ok")
@@ -52,3 +52,23 @@ export function isNonPublicIp(ip: string): boolean {
    || inRange("255.255.255.255/32")
  )
 }
+
+const PEER_WELL_KNOWN_PORTS = new Set([80, 443, 53, 853])
+
+/**
+ * Интернет-сторона потока: у IPFIX сервис часто в src (Google:443 → RFC1918:ephemeral).
+ * Классифицировать этот IP, не слепой dst.
+ */
+export function pickInternetPeer(src: string, dst: string, srcPort: number, dstPort: number): string {
+  const srcPub = !isNonPublicIp(src)
+  const dstPub = !isNonPublicIp(dst)
+  if (srcPub && !dstPub) return src
+  if (dstPub && !srcPub) return dst
+  if (srcPub && dstPub) {
+    const srcWk = PEER_WELL_KNOWN_PORTS.has(srcPort)
+    const dstWk = PEER_WELL_KNOWN_PORTS.has(dstPort)
+    if (srcWk && !dstWk) return src
+    if (dstWk && !srcWk) return dst
+  }
+  return dst
+}
@@ -0,0 +1,417 @@
+import assert from "node:assert/strict"
+import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js"
+import {
+  ingestParsedFlowsForServerForTests,
+  resetFlowRingsForTests,
+} from "./traffic-flow-ingest.js"
+import { buildFlowMapHops, resetFlowMapHopsCacheForTests } from "./traffic-flow-map-hops.js"
+import { seedFlowTopologyForTests, type FlowTopology } from "./traffic-flow-topology.js"
+import { disableCatalogFetchForTests, resetFlowCatalogForTests } from "./traffic-flow-classify.js"
+import {
+  disableRipeEnqueueForTests,
+  disableRipePersistForTests,
+  resetRipeCacheForTests,
+  seedRipeCacheForTests,
+} from "./traffic-flow-ripe.js"
+
+disableCatalogFetchForTests()
+resetFlowCatalogForTests()
+disableRipePersistForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+
+const topo: FlowTopology = {
+  clientIfaces: new Map([[7, new Set(["gre-client"])]]),
+  clientByIface: new Map([["7|gre-client", {
+    userId: "u1",
+    login: "alice",
+    name: "Alice",
+    serverId: 7,
+    interfaceName: "gre-client",
+  }]]),
+  enNodes: [{ id: 9, name: "NSK-EN", hosts: ["198.51.100.1"] }],
+  enHosts: new Set(["198.51.100.1"]),
+  jhHosts: new Set(["203.0.113.10"]),
+  wanIfaces: new Map([[3, new Set(["ether1-rt"])]]),
+  plane: {
+    clientIfaceNames: new Set(["gre-client"]),
+    enHosts: new Set(["198.51.100.1"]),
+    jhHosts: new Set(["203.0.113.10"]),
+  },
+}
+
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+seedFlowTopologyForTests(topo)
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "gre-client" },
+  { ".id": "*3", name: "gre-jh-en" },
+  { ".id": "*A", name: "wg-flow" },
+])
+rememberServerIfaces(3, [
+  { ".id": "*1", name: "ether1-rt" },
+])
+
+ingestParsedFlowsForServerForTests(7, [
+  {
+    src: "10.100.1.17",
+    dst: "8.8.8.8",
+    proto: 6,
+    srcPort: 51234,
+    dstPort: 443,
+    bytes: 12_000,
+    packets: 10,
+    inIface: "2",
+    outIface: "3",
+    nextHop: "198.51.100.1",
+  },
+  {
+    src: "203.0.113.10",
+    dst: "198.51.100.1",
+    proto: 47,
+    srcPort: 0,
+    dstPort: 0,
+    bytes: 5_000_000,
+    packets: 4000,
+    inIface: "3",
+    outIface: "3",
+  },
+  {
+    src: "10.100.1.17",
+    dst: "10.100.1.18",
+    proto: 6,
+    srcPort: 50000,
+    dstPort: 443,
+    bytes: 8000,
+    packets: 8,
+    inIface: "2",
+    outIface: "2",
+  },
+  {
+    src: "10.255.254.1",
+    dst: "10.255.254.2",
+    proto: 17,
+    srcPort: 4739,
+    dstPort: 2055,
+    bytes: 400,
+    packets: 2,
+    inIface: "10",
+    outIface: "",
+  },
+])
+ingestParsedFlowsForServerForTests(3, [
+  {
+    src: "192.168.1.10",
+    dst: "8.8.4.4",
+    proto: 6,
+    srcPort: 40000,
+    dstPort: 443,
+    bytes: 3000,
+    packets: 4,
+    inIface: "1",
+    outIface: "1",
+  },
+])
+
+try {
+  resetFlowMapHopsCacheForTests()
+  const def = buildFlowMapHops({ minutes: 5 })
+  assert.equal(def.excludeOverlayApplied, true)
+  assert.equal(def.excludeMeshApplied, true)
+  assert.equal(def.dedupApplied, true)
+  assert.equal(def.windowSec, 300)
+
+  const payloadGre = def.hops.find((h) => h.kind === "gre" && h.fromId === "7" && h.toId === "9")
+  assert.ok(payloadGre, "payload JH→EN hop")
+  assert.equal(payloadGre.bytes, 12_000)
+  assert.equal(payloadGre.bps, (12_000 * 8) / 300)
+  assert.equal(payloadGre.bpsFwd, (12_000 * 8) / 300)
+  assert.equal(payloadGre.iface, "gre-jh-en")
+
+  const greIface = def.hops.find((h) => h.kind === "iface" && h.iface === "gre-jh-en" && h.fromId === "7")
+  assert.ok(greIface)
+  assert.equal(greIface.bytes, 12_000)
+  assert.equal(greIface.bpsFwd, (12_000 * 8) / 300)
+
+  assert.ok(!def.hops.some((h) => h.bytes >= 5_000_000), "overlay GRE proto 47 excluded")
+  assert.ok(!def.hops.some((h) => h.iface === "wg-flow"), "mgmt wg-flow excluded")
+  const clientIngress = def.hops.find((h) => h.iface === "gre-client" && h.fromId === "7" && h.kind === "iface")
+  assert.ok(clientIngress, "payload ingress on client iface")
+  assert.equal(clientIngress.bytes, 12_000)
+
+  const wan = def.hops.find((h) => h.kind === "wan" && h.fromId === "3" && h.iface === "ether1-rt")
+  assert.ok(wan, "WAN hop from home-router")
+  assert.equal(wan.bytes, 3000)
+
+  resetFlowMapHopsCacheForTests()
+  const withAll = buildFlowMapHops({ minutes: 5, excludeOverlay: false, excludeMesh: false })
+  const overlayIface = withAll.hops.find((h) => h.iface === "gre-jh-en" && h.fromId === "7")
+  assert.ok(overlayIface && overlayIface.bytes >= 5_000_000)
+  const meshIface = withAll.hops.find((h) => h.iface === "gre-client" && h.fromId === "7" && h.kind === "iface")
+  assert.ok(meshIface && meshIface.bytes >= 20_000)
+} finally {
+  seedFlowTopologyForTests(null)
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
+console.log("traffic-flow-map-hops.test.ts: hops ok")
+
+function googleRipe() {
+  seedRipeCacheForTests({
+    prefix: "8.8.8.0/24",
+    asn: 15169,
+    country: "US",
+    lat: 37.4,
+    lng: -122.1,
+    holder: "GOOGLE",
+    ok: true,
+    fetchedAt: Date.now(),
+  })
+}
+
+function payloadFlow(dst: string, bytes: number) {
+  return {
+    src: "10.100.1.17",
+    dst,
+    proto: 6,
+    srcPort: 51234,
+    dstPort: 443,
+    bytes,
+    packets: Math.max(1, Math.round(bytes / 1200)),
+    inIface: "2",
+    outIface: "3",
+    nextHop: "198.51.100.1",
+  }
+}
+
+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", 600),
+  payloadFlow("203.0.113.50", 9400),
+])
+try {
+  resetFlowMapHopsCacheForTests()
+  const six = buildFlowMapHops({ minutes: 5, minSharePct: 5 })
+  assert.equal(six.totalBytes, 10_000)
+  const google = six.services?.find((s) => s.id === "svc:google")
+  assert.ok(google, "Google ≥ 5%")
+  assert.ok(google.share >= 0.05)
+  const googleEdge = six.serviceEdges?.find((e) => e.toId === "svc:google" && e.fromId === "9")
+  assert.ok(googleEdge)
+  assert.equal(googleEdge.clientName, "Alice")
+} finally {
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+}
+
+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", 400),
+  payloadFlow("203.0.113.50", 9600),
+])
+try {
+  resetFlowMapHopsCacheForTests()
+  const four = buildFlowMapHops({ minutes: 5, minSharePct: 5 })
+  assert.equal(four.totalBytes, 10_000)
+  assert.ok(!(four.services ?? []).some((s) => s.id === "svc:google"), "Google < 5% hidden")
+  resetFlowMapHopsCacheForTests()
+  const off = buildFlowMapHops({ minutes: 5, minSharePct: 0 })
+  assert.ok(off.services?.some((s) => s.id === "svc:google"), "порог 0 показывает Google 4%")
+} finally {
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+}
+
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+seedFlowTopologyForTests(topo)
+rememberServerIfaces(7, [
+  { ".id": "*2", name: "gre-client" },
+  { ".id": "*3", name: "gre-jh-en" },
+])
+ingestParsedFlowsForServerForTests(7, [
+  {
+    src: "203.0.113.10",
+    dst: "198.51.100.1",
+    proto: 47,
+    srcPort: 0,
+    dstPort: 0,
+    bytes: 9_000,
+    packets: 90,
+    inIface: "3",
+    outIface: "3",
+  },
+  payloadFlow("203.0.113.50", 1000),
+])
+try {
+  resetFlowMapHopsCacheForTests()
+  const greOnly = buildFlowMapHops({ minutes: 5, excludeOverlay: false, minSharePct: 0 })
+  assert.ok(!(greOnly.services ?? []).some((s) => s.label === "GRE"), "GRE is not a destination service")
+} 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" },
+])
+ingestParsedFlowsForServerForTests(7, [
+  {
+    src: "173.194.151.65",
+    dst: "10.200.100.53",
+    proto: 6,
+    srcPort: 443,
+    dstPort: 57182,
+    bytes: 9_000,
+    packets: 90,
+    inIface: "2",
+    outIface: "3",
+    nextHop: "198.51.100.1",
+  },
+  {
+    src: "104.18.35.51",
+    dst: "10.200.100.53",
+    proto: 6,
+    srcPort: 443,
+    dstPort: 53880,
+    bytes: 1_000,
+    packets: 10,
+    inIface: "2",
+    outIface: "3",
+    nextHop: "198.51.100.1",
+  },
+])
+try {
+  resetFlowMapHopsCacheForTests()
+  const rev = buildFlowMapHops({ minutes: 5, minSharePct: 0 })
+  assert.ok(rev.services?.some((s) => s.id === "svc:google"), "реверс Google:443 → 10.x")
+  assert.ok(rev.services?.some((s) => s.id === "svc:cloudflare"), "реверс Cloudflare:443 → 10.x")
+  assert.ok(rev.serviceEdges?.some((e) => e.toId === "svc:google" && e.fromId === "9"))
+} finally {
+  seedFlowTopologyForTests(null)
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+seedFlowTopologyForTests(topo)
+rememberServerIfaces(7, [
+  { ".id": "*1", name: "SWE-VEESP" },
+  { ".id": "*2", name: "gre-client" },
+  { ".id": "*3", name: "gre-jh-en" },
+])
+ingestParsedFlowsForServerForTests(7, [
+  payloadFlow("8.8.8.8", 500),
+  {
+    src: "173.194.151.65",
+    dst: "10.200.100.53",
+    proto: 6,
+    srcPort: 443,
+    dstPort: 57182,
+    bytes: 8_000,
+    packets: 80,
+    inIface: "1",
+    outIface: "1",
+    nextHop: "",
+  },
+])
+try {
+  resetFlowMapHopsCacheForTests()
+  const wan = buildFlowMapHops({ minutes: 5, minSharePct: 0 })
+  const googleEdge = wan.serviceEdges?.find((e) => e.toId === "svc:google")
+  assert.ok(googleEdge, "Google с WAN JH")
+  assert.equal(googleEdge.fromId, "9", "якорь на EN, не на JH")
+  assert.ok(!(wan.serviceEdges ?? []).some((e) => e.fromId === "7"), "нет пунктира с JH")
+  const viaGre = wan.serviceEdges?.find((e) => e.toId === "svc:google")
+  assert.ok(viaGre?.clients?.some((c) => c.name === "Alice") || viaGre?.clientName === "Alice")
+} finally {
+  seedFlowTopologyForTests(null)
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
+resetFlowRingsForTests()
+resetIfaceCacheForTests()
+resetRipeCacheForTests()
+disableRipeEnqueueForTests()
+seedFlowTopologyForTests(topo)
+rememberServerIfaces(7, [
+  { ".id": "*1", name: "SWE-VEESP" },
+])
+ingestParsedFlowsForServerForTests(7, [
+  {
+    src: "173.194.151.65",
+    dst: "10.200.100.53",
+    proto: 6,
+    srcPort: 443,
+    dstPort: 57182,
+    bytes: 9_000,
+    packets: 80,
+    inIface: "1",
+    outIface: "1",
+    nextHop: "",
+  },
+])
+try {
+  resetFlowMapHopsCacheForTests()
+  const wanOnly = buildFlowMapHops({ minutes: 5, minSharePct: 0 })
+  const googleEdge = wanOnly.serviceEdges?.find((e) => e.toId === "svc:google")
+  assert.ok(googleEdge, "Google WAN без GRE payload")
+  assert.equal(googleEdge.fromId, "9", "единственный EN, даже без nextHop")
+  assert.ok(googleEdge.bps > 0, "скорость на hop EN→сервис")
+  assert.ok(!(wanOnly.serviceEdges ?? []).some((e) => e.fromId === "7"), "нет пунктира с JH")
+  const googlePath = wanOnly.servicePaths?.find((p) => p.serviceId === "svc:google")
+  assert.ok(googlePath, "путь WAN Google")
+  assert.equal(googlePath.viaId, "7", "via = JH exporter")
+  assert.equal(googlePath.enId, "9", "якорь EN")
+  assert.ok(googlePath.bps > 0, "скорость на пути клиента")
+} finally {
+  seedFlowTopologyForTests(null)
+  resetFlowRingsForTests()
+  resetIfaceCacheForTests()
+  resetRipeCacheForTests()
+  resetFlowCatalogForTests()
+}
+
+console.log("traffic-flow-map-hops.test.ts: ok")
@@ -0,0 +1,521 @@
+import { eq } from "drizzle-orm"
+import type { FlowMapHop, FlowMapHopsDto, FlowMapService, FlowMapServiceEdge, FlowMapServicePath } from "@mmapp/contracts/traffic-flow"
+import { db } from "../db/index.js"
+import { userInterfaceBindings } from "../db/schema.js"
+import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js"
+import {
+  isNamedInternetService,
+  lookupBrand,
+  mapServiceNodeId,
+} from "./traffic-flow-brands.js"
+import { dedupFlowRowsMaxBytes } from "./traffic-flow-dedup.js"
+import { getFlowListenerState, listFlowRowsForWindow } from "./traffic-flow-ingest.js"
+import { resolveIfaceName } from "./traffic-flow-ifaces.js"
+import { classifyFlowPlane, shouldKeepPlane } from "./traffic-flow-planes.js"
+import { pickInternetPeer } from "./traffic-flow-ip.js"
+import { lookupRipeCached, type FlowIpMeta } from "./traffic-flow-ripe.js"
+import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js"
+import { loadFlowTopology, resolveClient, resolveEn, getServerCatalog } from "./traffic-flow-topology.js"
+import { flowDataEpoch } from "./traffic-flow-engine.js"
+
+export const DEFAULT_MAP_SERVICE_MIN_SHARE_PCT = 5
+export const MAP_SERVICE_NODE_CAP = 20
+const HOPS_CACHE_TTL_MS = 2000
+
+export interface FlowMapHopsQuery {
+  minutes: number
+  serverId?: number
+  userId?: string
+  iface?: string
+  dedup?: boolean
+  excludeMesh?: boolean
+  excludeOverlay?: boolean
+  /** Переопределение порога (тесты). Иначе из настроек NetFlow. */
+  minSharePct?: number
+}
+
+interface HopAcc {
+  fromId: string
+  fromLabel: string
+  toId: string
+  toLabel: string
+  kind: FlowMapHop["kind"]
+  iface?: string
+  bytes: number
+  bytesFwd: number
+  bytesRev: number
+}
+
+interface ClientAcc {
+  name: string
+  bytes: number
+}
+
+interface FromAcc {
+  bytes: number
+  clients: Map<string, ClientAcc>
+}
+
+interface DstAcc {
+  bytes: number
+  proto: number
+  dstPort: number
+  srcPort: number
+  fromBytes: Map<string, FromAcc>
+}
+
+function bumpClient(clients: Map<string, ClientAcc>, bytes: number, client: { userId: string; name: string } | null): void {
+  const id = client?.userId || "—"
+  const name = client?.name || "—"
+  const prev = clients.get(id)
+  if (prev) {
+    prev.bytes += bytes
+    return
+  }
+  clients.set(id, { name, bytes })
+}
+
+function bumpFrom(acc: DstAcc, exporterId: string, bytes: number, client: { userId: string; name: string } | null): void {
+  const prev = acc.fromBytes.get(exporterId)
+  if (prev) {
+    prev.bytes += bytes
+    bumpClient(prev.clients, bytes, client)
+    return
+  }
+  const clients = new Map<string, ClientAcc>()
+  bumpClient(clients, bytes, client)
+  acc.fromBytes.set(exporterId, { bytes, clients })
+}
+
+let hopsCache: { key: string; at: number; dto: FlowMapHopsDto } | null = null
+
+export function resetFlowMapHopsCacheForTests(): void {
+  hopsCache = null
+}
+
+export function clampMapServiceMinSharePct(n: unknown): number {
+  const v = typeof n === "number" ? n : Number(n)
+  if (!Number.isFinite(v)) return DEFAULT_MAP_SERVICE_MIN_SHARE_PCT
+  return Math.min(100, Math.max(0, v))
+}
+
+function hopsQueryKey(q: FlowMapHopsQuery, minSharePct: number): string {
+  return JSON.stringify({
+    epoch: flowDataEpoch(),
+    minutes: q.minutes,
+    serverId: q.serverId ?? null,
+    userId: q.userId ?? null,
+    iface: q.iface ?? null,
+    dedup: q.dedup !== false,
+    excludeMesh: q.excludeMesh !== false,
+    excludeOverlay: q.excludeOverlay !== false,
+    minSharePct,
+  })
+}
+
+function userIfaceAllow(userId: string): Map<number, Set<string>> | null {
+  if (!userId) return null
+  const binds = db.select().from(userInterfaceBindings).where(eq(userInterfaceBindings.userId, userId)).all()
+  const allow = new Map<number, Set<string>>()
+  for (const b of binds) {
+    const set = allow.get(b.serverId) ?? new Set<string>()
+    set.add(b.interfaceName)
+    allow.set(b.serverId, set)
+  }
+  return allow
+}
+
+function ifaceUsable(name: string): boolean {
+  return Boolean(name) && name !== "—"
+}
+
+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
+  const addRev = dir === "rev" || dir === "both" ? bytes : 0
+  if (prev) {
+    prev.bytes += bytes
+    prev.bytesFwd += addFwd
+    prev.bytesRev += addRev
+    if (seed.iface && !prev.iface) prev.iface = seed.iface
+    return
+  }
+  acc.set(key, {
+    ...seed,
+    bytes,
+    bytesFwd: addFwd,
+    bytesRev: addRev,
+  })
+}
+
+function toHop(a: HopAcc, windowSec: number): FlowMapHop {
+  return {
+    fromId: a.fromId,
+    fromLabel: a.fromLabel,
+    toId: a.toId,
+    toLabel: a.toLabel,
+    kind: a.kind,
+    ...(a.iface ? { iface: a.iface } : {}),
+    bytes: a.bytes,
+    bps: (a.bytes * 8) / windowSec,
+    bpsFwd: (a.bytesFwd * 8) / windowSec,
+    bpsRev: (a.bytesRev * 8) / windowSec,
+  }
+}
+
+/** Имя бренда без каталога EvoBGP — только ASN/CIDR кэш + proto. */
+function classifyMapDstLite(
+  dst: string,
+  proto: number,
+  dstPort: number,
+  srcPort: number,
+  ripe: FlowIpMeta | null,
+): { service: string; category: string } | null {
+  if (proto === 47 || proto === 50) return null
+  const app = applicationName(proto, dstPort, srcPort)
+  if (app === "WireGuard" || app === "DNS" || app === "SSH" || app === "BGP") return null
+  if (/youtube/i.test(ripe?.holder ?? "")) {
+    return { service: "YouTube", category: "Видео / стриминг" }
+  }
+  const brand = lookupBrand(dst, ripe?.asn ?? 0)
+  if (!brand || !isNamedInternetService(brand.service, brand.category)) return null
+  return brand
+}
+
+function resolveMinSharePct(q: FlowMapHopsQuery): number {
+  if (q.minSharePct != null) return clampMapServiceMinSharePct(q.minSharePct)
+  try {
+    const row = getTrafficFlowSettingsRow() as { mapServiceMinSharePct?: number }
+    return clampMapServiceMinSharePct(row.mapServiceMinSharePct ?? DEFAULT_MAP_SERVICE_MIN_SHARE_PCT)
+  } catch {
+    return DEFAULT_MAP_SERVICE_MIN_SHARE_PCT
+  }
+}
+
+function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number): FlowMapHopsDto {
+  const windowSec = Math.max(60, q.minutes * 60)
+  const raw = listFlowRowsForWindow(q.minutes)
+  const allow = q.userId ? userIfaceAllow(q.userId) : null
+  const catalog = getServerCatalog()
+  const nameById = new Map([...catalog.byId].map(([id, s]) => [id, s.name]))
+  const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : ""
+  const wantDedup = q.dedup !== false && !ifaceFilter
+  const excludeMesh = q.excludeMesh !== false
+  const excludeOverlay = q.excludeOverlay !== false
+  const topo = loadFlowTopology()
+
+  const matched = []
+  for (const r of raw) {
+    const resolved = resolveIfaceName(r.serverId, r.inIface)
+    const outResolved = resolveIfaceName(r.serverId, r.outIface)
+    if (!flowRowMatchesFilter(r, resolved.name, q, allow)) continue
+    const plane = classifyFlowPlane({
+      src: r.src,
+      dst: r.dst,
+      proto: r.proto,
+      srcPort: r.srcPort,
+      dstPort: r.dstPort,
+      inIface: resolved.name,
+      outIface: outResolved.name,
+    }, topo.plane)
+    if (!shouldKeepPlane(plane, { excludeMesh, excludeOverlay })) continue
+    matched.push(r)
+  }
+
+  const working = wantDedup ? dedupFlowRowsMaxBytes(matched) : matched
+  const hops = new Map<string, HopAcc>()
+  const dstAcc = new Map<string, DstAcc>()
+  const jhToEn = new Map<number, number>()
+  const enIds = new Set(topo.enNodes.map((n) => n.id))
+  let totalBytes = 0
+
+  for (const r of working) {
+    const inRes = resolveIfaceName(r.serverId, r.inIface)
+    const outRes = resolveIfaceName(r.serverId, r.outIface)
+    const inName = inRes.name
+    const outName = outRes.name
+    const fromId = String(r.serverId)
+    const fromLabel = nameById.get(r.serverId) ?? fromId
+    const wanSet = topo.wanIfaces.get(r.serverId)
+
+    const inOk = ifaceUsable(inName)
+    const outOk = ifaceUsable(outName)
+    const sameIface = inOk && outOk && inName.toLowerCase() === outName.toLowerCase()
+    if (sameIface) {
+      bump(hops, `iface|${fromId}|${inName.toLowerCase()}`, {
+        fromId,
+        fromLabel,
+        toId: "",
+        toLabel: "",
+        kind: "iface",
+        iface: inName,
+      }, r.bytes, "fwd")
+    } else {
+      if (inOk) {
+        bump(hops, `iface|${fromId}|${inName.toLowerCase()}`, {
+          fromId,
+          fromLabel,
+          toId: "",
+          toLabel: "",
+          kind: "iface",
+          iface: inName,
+        }, r.bytes, "rev")
+      }
+      if (outOk) {
+        bump(hops, `iface|${fromId}|${outName.toLowerCase()}`, {
+          fromId,
+          fromLabel,
+          toId: "",
+          toLabel: "",
+          kind: "iface",
+          iface: outName,
+        }, r.bytes, "fwd")
+      }
+    }
+
+    const enOut = ifaceUsable(outName) ? resolveEn(topo, r.nextHop, outName) : null
+    const enIn = ifaceUsable(inName) ? resolveEn(topo, "", inName) : null
+    const en = (enOut && enOut.id !== r.serverId ? enOut : null)
+      ?? (enIn && enIn.id !== r.serverId ? enIn : null)
+    if (en) {
+      jhToEn.set(r.serverId, en.id)
+      const toId = String(en.id)
+      const dir: "fwd" | "rev" = enOut && enOut.id === en.id ? "fwd" : "rev"
+      const greIface = dir === "fwd" && ifaceUsable(outName) ? outName : (ifaceUsable(inName) ? inName : undefined)
+      bump(hops, `gre|${fromId}|${toId}`, {
+        fromId,
+        fromLabel,
+        toId,
+        toLabel: en.name,
+        kind: "gre",
+        iface: greIface,
+      }, r.bytes, dir)
+    }
+
+    if (wanSet?.size) {
+      if (ifaceUsable(inName) && wanSet.has(inName)) {
+        bump(hops, `wan|${fromId}|${inName.toLowerCase()}`, {
+          fromId,
+          fromLabel,
+          toId: "",
+          toLabel: "",
+          kind: "wan",
+          iface: inName,
+        }, r.bytes, "rev")
+      }
+      if (ifaceUsable(outName) && wanSet.has(outName) && outName.toLowerCase() !== inName.toLowerCase()) {
+        bump(hops, `wan|${fromId}|${outName.toLowerCase()}`, {
+          fromId,
+          fromLabel,
+          toId: "",
+          toLabel: "",
+          kind: "wan",
+          iface: outName,
+        }, r.bytes, "fwd")
+      }
+    }
+
+    totalBytes += r.bytes
+    const peer = pickInternetPeer(r.src, r.dst, r.srcPort, r.dstPort)
+    const client = resolveClient(topo, r.serverId, inName)
+    const prevDst = dstAcc.get(peer)
+    if (prevDst) {
+      prevDst.bytes += r.bytes
+      bumpFrom(prevDst, String(r.serverId), r.bytes, client)
+    } else {
+      const acc: DstAcc = {
+        bytes: r.bytes,
+        proto: r.proto,
+        dstPort: r.dstPort,
+        srcPort: r.srcPort,
+        fromBytes: new Map(),
+      }
+      bumpFrom(acc, String(r.serverId), r.bytes, client)
+      dstAcc.set(peer, acc)
+    }
+  }
+
+  const svcTotals = new Map<string, { label: string; category: string; bytes: number }>()
+  const svcEdges = new Map<string, {
+    fromId: string
+    toId: string
+    bytes: number
+    bytesFwd: number
+    bytesRev: number
+    clients: Map<string, string>
+  }>()
+  const svcPaths = new Map<string, {
+    clientId: string
+    clientName: string
+    viaId: string
+    viaName: string
+    enId: string
+    enName: string
+    serviceId: string
+    bytes: number
+  }>()
+
+  for (const h of hops.values()) {
+    if (h.kind !== "gre" || !h.toId) continue
+    const from = Number(h.fromId)
+    const to = Number(h.toId)
+    if (!Number.isFinite(from) || !Number.isFinite(to)) continue
+    if (enIds.has(to) && !enIds.has(from)) jhToEn.set(from, to)
+  }
+
+  const soleEnId = topo.enNodes.length === 1 ? String(topo.enNodes[0]!.id) : null
+
+  function anchorEnId(exporterId: string): string | null {
+    const n = Number(exporterId)
+    if (enIds.has(n)) return exporterId
+    const mapped = jhToEn.get(n)
+    if (mapped != null) return String(mapped)
+    if (soleEnId) return soleEnId
+    return null
+  }
+
+  function nodeName(id: string): string {
+    const n = Number(id)
+    if (Number.isFinite(n)) {
+      const fromDb = nameById.get(n)
+      if (fromDb) return fromDb
+    }
+    const en = topo.enNodes.find((node) => String(node.id) === id)
+    if (en?.name) return en.name
+    return id
+  }
+
+  for (const [dst, acc] of dstAcc) {
+    const ripe = lookupRipeCached(dst)
+    const classified = classifyMapDstLite(dst, acc.proto, acc.dstPort, acc.srcPort, ripe)
+    if (!classified) continue
+    const toId = mapServiceNodeId(classified.service)
+    const prevSvc = svcTotals.get(toId)
+    if (prevSvc) prevSvc.bytes += acc.bytes
+    else svcTotals.set(toId, { label: classified.service, category: classified.category, bytes: acc.bytes })
+    for (const [exporterId, from] of acc.fromBytes) {
+      const fromId = anchorEnId(exporterId)
+      if (!fromId) continue
+      const edgeKey = `${fromId}|${toId}`
+      const prevEdge = svcEdges.get(edgeKey)
+      const namedClients = new Map<string, string>()
+      for (const [id, c] of from.clients) {
+        if (id !== "—") namedClients.set(id, c.name)
+      }
+      if (prevEdge) {
+        prevEdge.bytes += from.bytes
+        prevEdge.bytesFwd += from.bytes
+        for (const [id, name] of namedClients) prevEdge.clients.set(id, name)
+      } else {
+        svcEdges.set(edgeKey, {
+          fromId,
+          toId,
+          bytes: from.bytes,
+          bytesFwd: from.bytes,
+          bytesRev: 0,
+          clients: namedClients,
+        })
+      }
+      const enName = nodeName(fromId)
+      const viaName = nodeName(exporterId)
+      for (const [clientId, c] of from.clients) {
+        const pathKey = `${clientId}|${exporterId}|${fromId}|${toId}`
+        const prevPath = svcPaths.get(pathKey)
+        if (prevPath) {
+          prevPath.bytes += c.bytes
+        } else {
+          svcPaths.set(pathKey, {
+            clientId,
+            clientName: c.name,
+            viaId: exporterId,
+            viaName,
+            enId: fromId,
+            enName,
+            serviceId: toId,
+            bytes: c.bytes,
+          })
+        }
+      }
+    }
+  }
+
+  const minShare = minSharePct / 100
+  let services: FlowMapService[] = [...svcTotals.entries()]
+    .map(([id, s]) => ({
+      id,
+      label: s.label,
+      category: s.category,
+      bytes: s.bytes,
+      bps: (s.bytes * 8) / windowSec,
+      share: totalBytes > 0 ? s.bytes / totalBytes : 0,
+    }))
+    .sort((a, b) => b.bytes - a.bytes)
+  if (minSharePct > 0) {
+    services = services.filter((s) => s.share >= minShare)
+  }
+  services = services.slice(0, MAP_SERVICE_NODE_CAP)
+  const keepSvc = new Set(services.map((s) => s.id))
+  const serviceEdges: FlowMapServiceEdge[] = [...svcEdges.values()]
+    .filter((e) => keepSvc.has(e.toId))
+    .map((e) => {
+      const clients = [...e.clients.entries()].map(([id, name]) => ({ id, name }))
+      const first = clients[0]
+      return {
+        fromId: e.fromId,
+        toId: e.toId,
+        bytes: e.bytes,
+        bps: (e.bytes * 8) / windowSec,
+        bpsFwd: (e.bytesFwd * 8) / windowSec,
+        bpsRev: (e.bytesRev * 8) / windowSec,
+        ...(first ? { clientId: first.id, clientName: first.name } : {}),
+        ...(clients.length ? { clients } : {}),
+      }
+    })
+    .sort((a, b) => b.bytes - a.bytes)
+
+  const servicePaths: FlowMapServicePath[] = [...svcPaths.values()]
+    .filter((p) => keepSvc.has(p.serviceId))
+    .map((p) => ({
+      clientId: p.clientId,
+      clientName: p.clientName,
+      viaId: p.viaId,
+      viaName: p.viaName,
+      enId: p.enId,
+      enName: p.enName,
+      serviceId: p.serviceId,
+      bytes: p.bytes,
+      bps: (p.bytes * 8) / windowSec,
+    }))
+    .sort((a, b) => b.bps - a.bps)
+
+  const listener = getFlowListenerState()
+  return {
+    hops: [...hops.values()]
+      .map((a) => toHop(a, windowSec))
+      .sort((a, b) => a.bytes === b.bytes ? 0 : b.bytes - a.bytes),
+    live: listener.bound,
+    rangeMinutes: q.minutes,
+    windowSec,
+    totalBytes,
+    services,
+    serviceEdges,
+    servicePaths,
+    mapServiceMinSharePct: minSharePct,
+    dedupApplied: wantDedup,
+    excludeMeshApplied: excludeMesh,
+    excludeOverlayApplied: excludeOverlay,
+  }
+}
+
+/** Hop-rates для карты сети: те же фильтры, что у общего NetFlow (dedup / mesh / overlay). */
+export function buildFlowMapHops(q: FlowMapHopsQuery): FlowMapHopsDto {
+  const minSharePct = resolveMinSharePct(q)
+  const key = hopsQueryKey(q, minSharePct)
+  const now = Date.now()
+  if (hopsCache && hopsCache.key === key && now - hopsCache.at < HOPS_CACHE_TTL_MS) {
+    return hopsCache.dto
+  }
+  const dto = buildFlowMapHopsUncached(q, minSharePct)
+  hopsCache = { key, at: now, dto }
+  return dto
+}
@@ -21,6 +21,7 @@ import {
 } from "./traffic-flow-settings.js"
 import { refreshFlowExporterMap, startTrafficFlowListener } from "./traffic-flow-ingest.js"
 import { listTrafficFlowHostFiles } from "./traffic-flow-host-files.js"
+import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js"

 const IFACE_NAME = "wg-flow"
 const JH_LISTEN_PORT = 13232
@@ -288,6 +289,7 @@ export async function applyFlowOverlay(
      mgmtTunnelIp: address,
      updatedAt: new Date().toISOString(),
    }).where(eq(servers.id, server.id)).run()
+    invalidateFlowCatalogCache()

    upsertHostPeer({
      serverId: server.id,
@@ -43,8 +43,8 @@ try {
  `).run(serverId)
  sqliteDatabase.prepare(`
    INSERT INTO flow_ip_meta (prefix, asn, country, holder, ok, fetched_at)
-    VALUES ('8.8.8.0/24', 15169, 'US', 'Google', 1, '2026-01-01T00:00:00.000Z')
-  `).run()
+    VALUES ('8.8.8.0/24', 15169, 'US', 'Google', 1, ?)
+  `).run(new Date().toISOString())
  sqliteDatabase.prepare(`
    UPDATE traffic_flow_settings SET packets_received = 42, last_exporter_ip = '10.255.254.3' WHERE id = 1
  `).run()
@@ -7,6 +7,7 @@ import {
  lookupRipeCached,
  resetRipeCacheForTests,
  ripeFetchCountForTests,
+  ripeLastCandidateCountForTests,
  seedRipeCacheForTests,
  setRipeFetchForTests,
 } from "./traffic-flow-ripe.js"
@@ -100,4 +101,36 @@ await flushRipeQueueForTests()
 assert.equal(lookupRipeCached("1.0.0.1")?.country, "US")
 assert.equal(lookupRipeCached("1.0.0.1")?.asn, 13335)

+resetRipeCacheForTests()
+disableRipePersistForTests()
+for (let i = 0; i < 3000; i++) {
+  const o2 = Math.floor(i / 256)
+  const o3 = i % 256
+  seedRipeCacheForTests({
+    prefix: `203.${o2}.${o3}.0/24`,
+    asn: 64500,
+    country: "NL",
+    lat: null,
+    lng: null,
+    holder: "NOISE",
+    ok: true,
+    fetchedAt: Date.now(),
+  })
+}
+seedRipeCacheForTests({
+  prefix: "8.8.8.0/24",
+  asn: 15169,
+  country: "US",
+  lat: null,
+  lng: null,
+  holder: "GOOGLE",
+  ok: true,
+  fetchedAt: Date.now(),
+})
+assert.equal(lookupRipeCached("8.8.8.8")?.asn, 15169)
+assert.ok(
+  ripeLastCandidateCountForTests() < 8,
+  `index should not scan all prefixes, got ${ripeLastCandidateCountForTests()}`,
+)
+
 console.log("traffic-flow-ripe.test.ts: ok")
@@ -1,5 +1,5 @@
 import { sqliteDatabase } from "../db/index.js"
-import { ipInCidrV4, ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
+import { ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
 import { resolveRipeCountry } from "./traffic-flow-brands.js"

 export interface FlowIpMeta {
@@ -28,6 +28,18 @@ const queue: string[] = []
 const queued = new Set<string>()
 const recentFetches: number[] = []

+interface RipeIndexed {
+  entry: FlowIpMeta
+  net: number
+  mask: number
+  prefixLen: number
+}
+
+/** /24 → кандидаты с prefixLen ≥ 24. Более широкие префиксы — в `wideIndex`. */
+const v24Index = new Map<number, RipeIndexed[]>()
+const wideIndex: RipeIndexed[] = []
+let lastCandidateCount = 0
+
 let persistEnabled = true
 let enqueueEnabled = true
 let loaded = false
@@ -50,6 +62,9 @@ export function resetRipeCacheForTests(): void {
  queue.length = 0
  queued.clear()
  recentFetches.length = 0
+  v24Index.clear()
+  wideIndex.length = 0
+  lastCandidateCount = 0
  loaded = persistEnabled ? false : true
  workerRunning = false
  fetchCount = 0
@@ -58,10 +73,15 @@ export function resetRipeCacheForTests(): void {
 }

 export function seedRipeCacheForTests(entry: FlowIpMeta): void {
-  mem.set(entry.prefix, { ...entry })
+  remember(entry)
  loaded = true
 }

+/** Сколько CIDR смотрели в последнем lookup (для теста индекса /24). */
+export function ripeLastCandidateCountForTests(): number {
+  return lastCandidateCount
+}
+
 export function setRipeFetchForTests(fn: typeof fetch): void {
  fetchImpl = fn
  fetchCount = 0
@@ -87,6 +107,48 @@ function isFresh(entry: FlowIpMeta): boolean {
  return Date.now() - entry.fetchedAt < ttlMs(entry.ok)
 }

+function unindexPrefix(prefix: string): void {
+  const parsed = parseCidrV4(prefix)
+  if (!parsed) return
+  if (parsed.prefixLen >= 24) {
+    const key = parsed.net >>> 8
+    const list = v24Index.get(key)
+    if (!list) return
+    const next = list.filter((row) => row.entry.prefix !== prefix)
+    if (next.length) v24Index.set(key, next)
+    else v24Index.delete(key)
+    return
+  }
+  const idx = wideIndex.findIndex((row) => row.entry.prefix === prefix)
+  if (idx >= 0) wideIndex.splice(idx, 1)
+}
+
+function indexEntry(entry: FlowIpMeta): void {
+  const parsed = parseCidrV4(entry.prefix)
+  if (!parsed) return
+  const row: RipeIndexed = {
+    entry,
+    net: parsed.net,
+    mask: parsed.mask,
+    prefixLen: parsed.prefixLen,
+  }
+  if (parsed.prefixLen >= 24) {
+    const key = parsed.net >>> 8
+    const list = v24Index.get(key)
+    if (list) list.push(row)
+    else v24Index.set(key, [row])
+    return
+  }
+  wideIndex.push(row)
+}
+
+function remember(entry: FlowIpMeta): void {
+  const prev = mem.get(entry.prefix)
+  if (prev) unindexPrefix(prev.prefix)
+  mem.set(entry.prefix, entry)
+  indexEntry(entry)
+}
+
 function loadSqlite(): void {
  if (loaded || !persistEnabled) {
    loaded = true
@@ -111,7 +173,7 @@ function loadSqlite(): void {
      const fetchedAt = Date.parse(r.fetched_at)
      const asn = Number(r.asn ?? 0) || 0
      const holder = r.holder || ""
-      mem.set(r.prefix, {
+      remember({
        prefix: r.prefix,
        asn,
        country: resolveRipeCountry(r.country || "", asn, holder) || "—",
@@ -177,6 +239,20 @@ function persistAsn(asn: number, holder: string): void {
  }
 }

+/** Удаляет просроченный RIPE-кэш с диска (hit 24h / negative 6h). */
+export function pruneRipeSqlite(nowMs = Date.now()): void {
+  if (!persistEnabled) return
+  try {
+    const hitCutoff = new Date(nowMs - HIT_TTL_MS).toISOString()
+    const negCutoff = new Date(nowMs - NEG_TTL_MS).toISOString()
+    sqliteDatabase.prepare(`DELETE FROM flow_ip_meta WHERE ok != 0 AND fetched_at < ?`).run(hitCutoff)
+    sqliteDatabase.prepare(`DELETE FROM flow_ip_meta WHERE ok = 0 AND fetched_at < ?`).run(negCutoff)
+    sqliteDatabase.prepare(`DELETE FROM flow_asn_meta WHERE fetched_at < ?`).run(hitCutoff)
+  } catch {
+    /* table may not exist in isolated tests */
+  }
+}
+
 function negative(prefix: string): FlowIpMeta {
  return {
    prefix,
@@ -193,20 +269,24 @@ function negative(prefix: string): FlowIpMeta {
 export function lookupRipeCached(ip: string): FlowIpMeta | null {
  loadSqlite()
  const trimmed = String(ip ?? "").trim()
+  lastCandidateCount = 0
  if (!trimmed) return null
  if (isNonPublicIp(trimmed)) {
    return negative(`${trimmed.includes(":") ? trimmed : trimmed}/32`)
  }
+  const addr = ipv4ToInt(trimmed)
+  if (addr == null) return null
+  const bucket = v24Index.get(addr >>> 8)
+  const candidates = bucket ? bucket.concat(wideIndex) : wideIndex
+  lastCandidateCount = candidates.length
  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
+  for (const row of candidates) {
+    if (!isFresh(row.entry)) continue
+    if (((addr & row.mask) >>> 0) !== row.net) continue
+    if (row.prefixLen > bestLen) {
+      best = row.entry
+      bestLen = row.prefixLen
    }
  }
  return best
@@ -316,13 +396,13 @@ async function resolveIp(ip: string): Promise<FlowIpMeta | null> {
        ok: Boolean(asn || country),
        fetchedAt: Date.now(),
      }
-      mem.set(prefix, entry)
+      remember(entry)
      persist(entry)
      return entry
    } catch {
      const prefix = `${ip}/32`
      const entry = negative(prefix)
-      mem.set(prefix, entry)
+      remember(entry)
      persist(entry)
      return entry
    } finally {
@@ -4,10 +4,42 @@ import { trafficFlowSettings } from "../db/schema.js"
 import type { FlowHostPeer, TrafficFlowSettingsDto, TrafficFlowSettingsPatch } from "@mmapp/contracts/traffic-flow"
 import { generateWireGuardKeyPair } from "./wg-keys.js"

+let settingsRowCache: ReturnType<typeof readSettingsRow> | null = null
+
 function nowIso() {
  return new Date().toISOString()
 }

+export function invalidateTrafficFlowSettingsCache(): void {
+  settingsRowCache = null
+}
+
+function readSettingsRow() {
+  return db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0]
+}
+
+export function getTrafficFlowSettingsRow() {
+  if (settingsRowCache) return settingsRowCache
+  const row = readSettingsRow()
+  if (row) {
+    settingsRowCache = row
+    return row
+  }
+  const now = nowIso()
+  db.insert(trafficFlowSettings).values({
+    id: 1,
+    enabled: false,
+    collectorIp: "10.255.254.1",
+    flowListenPort: 4739,
+    wgListenPort: 51821,
+    prefix: "10.255.254.0/24",
+    createdAt: now,
+    updatedAt: now,
+  }).run()
+  settingsRowCache = readSettingsRow()
+  return settingsRowCache!
+}
+
 function parsePeers(raw: string): FlowHostPeer[] {
  try {
    const parsed = JSON.parse(raw) as unknown
@@ -20,23 +52,6 @@ function parsePeers(raw: string): FlowHostPeer[] {
  }
 }

-export function getTrafficFlowSettingsRow() {
-  const row = db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0]
-  if (row) return row
-  const now = nowIso()
-  db.insert(trafficFlowSettings).values({
-    id: 1,
-    enabled: false,
-    collectorIp: "10.255.254.1",
-    flowListenPort: 4739,
-    wgListenPort: 51821,
-    prefix: "10.255.254.0/24",
-    createdAt: now,
-    updatedAt: now,
-  }).run()
-  return db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0]
-}
-
 export function toTrafficFlowSettingsDto(
  listener: { bound: boolean; address: string | null },
 ): TrafficFlowSettingsDto {
@@ -53,6 +68,7 @@ export function toTrafficFlowSettingsDto(
    hubServerId: row.hubServerId ?? null,
    retentionHours: row.retentionHours,
    topN: row.topN,
+    mapServiceMinSharePct: Number(row.mapServiceMinSharePct ?? 5),
    lastDatagramAt: row.lastDatagramAt ?? null,
    lastExporterIp: row.lastExporterIp ?? null,
    lastError: row.lastError || null,
@@ -75,8 +91,12 @@ export function updateTrafficFlowSettings(patch: TrafficFlowSettingsPatch) {
    hubServerId: patch.hubServerId === undefined ? row.hubServerId : patch.hubServerId,
    retentionHours: patch.retentionHours ?? row.retentionHours,
    topN: patch.topN ?? row.topN,
+    mapServiceMinSharePct: patch.mapServiceMinSharePct == null
+      ? row.mapServiceMinSharePct
+      : Math.min(100, Math.max(0, patch.mapServiceMinSharePct)),
    updatedAt: nowIso(),
  }).where(eq(trafficFlowSettings.id, 1)).run()
+  invalidateTrafficFlowSettingsCache()
  return getTrafficFlowSettingsRow()
 }

@@ -91,6 +111,7 @@ export function ensureHostKeys(): { publicKey: string; created: boolean } {
    hostPrivateKey: keys.privateKey,
    updatedAt: nowIso(),
  }).where(eq(trafficFlowSettings.id, 1)).run()
+  invalidateTrafficFlowSettingsCache()
  return { publicKey: keys.publicKey, created: true }
 }

@@ -102,6 +123,7 @@ export function upsertHostPeer(peer: FlowHostPeer) {
    peersJson: JSON.stringify(peers),
    updatedAt: nowIso(),
  }).where(eq(trafficFlowSettings.id, 1)).run()
+  invalidateTrafficFlowSettingsCache()
 }

 export function recordFlowPacket(exporterIp: string) {
@@ -112,6 +134,7 @@ export function recordFlowPacket(exporterIp: string) {
    packetsReceived: row.packetsReceived + 1,
    updatedAt: nowIso(),
  }).where(eq(trafficFlowSettings.id, 1)).run()
+  invalidateTrafficFlowSettingsCache()
 }

 export function recordFlowListenerError(message: string) {
@@ -119,6 +142,7 @@ export function recordFlowListenerError(message: string) {
    lastError: message,
    updatedAt: nowIso(),
  }).where(eq(trafficFlowSettings.id, 1)).run()
+  invalidateTrafficFlowSettingsCache()
 }

 export function enableTrafficFlowIngest() {
@@ -126,6 +150,7 @@ export function enableTrafficFlowIngest() {
    enabled: true,
    updatedAt: nowIso(),
  }).where(eq(trafficFlowSettings.id, 1)).run()
+  invalidateTrafficFlowSettingsCache()
 }

 export function listHostPeers(): FlowHostPeer[] {
@@ -140,4 +165,5 @@ export function resetFlowIngestCounters(): void {
    lastError: "",
    updatedAt: nowIso(),
  }).where(eq(trafficFlowSettings.id, 1)).run()
+  invalidateTrafficFlowSettingsCache()
 }
@@ -27,7 +27,44 @@ export interface FlowTopology {
  plane: PlaneTopology
 }

+export interface ServerCatalogEntry {
+  id: number
+  name: string
+  country: string
+  host: string
+  type: string
+  site: string
+}
+
+const CATALOG_TTL_MS = 5_000
+
 let seeded: FlowTopology | null = null
+let topologyCache: { at: number; topo: FlowTopology } | null = null
+let serverCatalogCache: { at: number; list: ServerCatalogEntry[]; byId: Map<number, ServerCatalogEntry> } | null = null
+
+export function invalidateFlowCatalogCache(): void {
+  topologyCache = null
+  serverCatalogCache = null
+}
+
+export function getServerCatalog(): { list: ServerCatalogEntry[]; byId: Map<number, ServerCatalogEntry> } {
+  const now = Date.now()
+  if (serverCatalogCache && now - serverCatalogCache.at < CATALOG_TTL_MS) {
+    return serverCatalogCache
+  }
+  const rows = db.select().from(servers).all()
+  const list: ServerCatalogEntry[] = rows.map((s) => ({
+    id: s.id,
+    name: s.name || s.host,
+    country: (s.country || "").toUpperCase() || "UN",
+    host: s.host,
+    type: s.type,
+    site: s.site || "—",
+  }))
+  const byId = new Map(list.map((s) => [s.id, s]))
+  serverCatalogCache = { at: now, list, byId }
+  return serverCatalogCache
+}

 function parseWanUplinks(raw: string): Array<{ iface?: string; ip?: string }> {
  try {
@@ -44,6 +81,8 @@ function ifaceKey(serverId: number, name: string): string {

 export function loadFlowTopology(): FlowTopology {
  if (seeded) return seeded
+  const now = Date.now()
+  if (topologyCache && now - topologyCache.at < CATALOG_TTL_MS) return topologyCache.topo
  const serverRows = db.select().from(servers).all()
  const users = db.select().from(appUsers).all()
  const binds = db.select().from(userInterfaceBindings).all()
@@ -82,7 +121,7 @@ export function loadFlowTopology(): FlowTopology {
      for (const h of hosts) jhHosts.add(h)
    }
  }
-  return {
+  const topo: FlowTopology = {
    clientIfaces,
    clientByIface,
    enNodes,
@@ -95,10 +134,13 @@ export function loadFlowTopology(): FlowTopology {
      jhHosts,
    },
  }
+  topologyCache = { at: Date.now(), topo }
+  return topo
 }

 export function seedFlowTopologyForTests(topo: FlowTopology | null): void {
  seeded = topo
+  invalidateFlowCatalogCache()
 }

 export function resolveClient(
@@ -0,0 +1,131 @@
+"use client"
+
+import type { ReactNode } from "react"
+
+function slug(label: string): string {
+  return label.trim().toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-+|-+$/g, "")
+}
+
+function GenericCloud({ size }: { size: number }) {
+  return (
+    <svg width={size} height={size} viewBox="0 0 24 24" aria-hidden>
+      <path
+        d="M7.5 18h9.2A4.3 4.3 0 0 0 21 13.8a4.2 4.2 0 0 0-3.7-4.2A6.1 6.1 0 0 0 6.2 11 3.8 3.8 0 0 0 3 14.7 3.7 3.7 0 0 0 6.8 18Z"
+        fill="#38bdf8"
+        opacity="0.92"
+      />
+    </svg>
+  )
+}
+
+function BrandSvg({ children, size }: { children: ReactNode; size: number }) {
+  return (
+    <svg width={size} height={size} viewBox="0 0 24 24" aria-hidden>
+      {children}
+    </svg>
+  )
+}
+
+export function ServiceBrandIcon({ label, size = 22 }: { label: string; size?: number }) {
+  switch (slug(label)) {
+    case "cloudflare":
+      return (
+        <BrandSvg size={size}>
+          <path d="M6.2 15.4h12.4c1.6 0 2.6-1.1 2.4-2.4-.2-1.4-1.4-2.1-2.8-2.1-.3-2.4-2.3-4.1-4.8-4.1-1.9 0-3.5 1-4.4 2.5-.4-.2-.9-.3-1.4-.3-1.7 0-3.1 1.3-3.2 3-.1 1.8 1.3 3.4 3.2 3.4Z" fill="#F38020" />
+        </BrandSvg>
+      )
+    case "google":
+      return (
+        <svg width={size} height={size} viewBox="0 0 48 48" aria-hidden>
+          <path fill="#FFC107" d="M43.6 20.1H42V20H24v8h11.3C33.7 32.7 29.3 36 24 36c-6.6 0-12-5.4-12-12s5.4-12 12-12c3.1 0 5.8 1.2 8 3l5.7-5.7C34 6.1 29.3 4 24 4 13 4 4 13 4 24s8.9 20 20 20c11 0 20-9 20-20 0-1.3-.1-2.7-.4-3.9z" />
+          <path fill="#FF3D00" d="M6.3 14.7 12.9 19.5C14.7 15.1 19 12 24 12c3.1 0 5.8 1.2 8 3l5.7-5.7C34 6.1 29.3 4 24 4 16.3 4 9.7 8.3 6.3 14.7z" />
+          <path fill="#4CAF50" d="M24 44c5.2 0 9.9-2 13.4-5.2l-6.2-5.2C29.2 35.1 26.7 36 24 36c-5.2 0-9.6-3.3-11.3-7.9l-6.5 5C9.5 39.6 16.2 44 24 44z" />
+          <path fill="#1976D2" d="M43.6 20.1H42V20H24v8h11.3c-.8 2.2-2.2 4.2-4.1 5.6l6.2 5.2C36.9 39.2 44 34 44 24c0-1.3-.1-2.7-.4-3.9z" />
+        </svg>
+      )
+    case "aws":
+    case "amazon":
+      return (
+        <BrandSvg size={size}>
+          <path d="M6.2 8.2 12 5.4l5.8 2.8v3.4L12 14.6 6.2 11.6Z" fill="#232F3E" />
+          <path d="M5.2 15.6c3.6 2.6 9.8 2.7 13.6 0" fill="none" stroke="#FF9900" strokeWidth="1.8" strokeLinecap="round" />
+        </BrandSvg>
+      )
+    case "steam":
+      return (
+        <BrandSvg size={size}>
+          <circle cx="12" cy="12" r="10" fill="#1b2838" />
+          <circle cx="8.2" cy="14.4" r="3.1" fill="#66c0f4" />
+          <circle cx="15.4" cy="9.2" r="3.6" fill="#c7d5e0" />
+          <circle cx="15.4" cy="9.2" r="1.5" fill="#1b2838" />
+        </BrandSvg>
+      )
+    case "blizzard":
+      return (
+        <BrandSvg size={size}>
+          <path d="M6 5h7.4c3 0 4.8 1.6 4.8 4.1 0 1.8-1 3.1-2.6 3.7 2 .5 3.2 2 3.2 4.1 0 2.8-2.1 4.6-5.6 4.6H6Z" fill="#00AEFF" />
+          <path d="M9.2 8.2h3.4c1.2 0 1.8.6 1.8 1.5s-.6 1.5-1.8 1.5H9.2Zm0 5.2h3.8c1.3 0 2 .6 2 1.6s-.7 1.6-2 1.6H9.2Z" fill="#06121f" />
+        </BrandSvg>
+      )
+    case "youtube":
+      return (
+        <BrandSvg size={size}>
+          <rect x="2" y="6" width="20" height="12" rx="3" fill="#FF0000" />
+          <path d="M10.2 9.2v5.6L15.6 12Z" fill="#fff" />
+        </BrandSvg>
+      )
+    case "netflix":
+      return (
+        <BrandSvg size={size}>
+          <path d="M6 3h3.2l5.6 18H11.6Z" fill="#E50914" />
+          <path d="M14.8 3H18v18h-3.2Z" fill="#B81D24" />
+        </BrandSvg>
+      )
+    case "microsoft":
+      return (
+        <BrandSvg size={size}>
+          <rect x="3" y="3" width="8" height="8" fill="#F25022" />
+          <rect x="13" y="3" width="8" height="8" fill="#7FBA00" />
+          <rect x="3" y="13" width="8" height="8" fill="#00A4EF" />
+          <rect x="13" y="13" width="8" height="8" fill="#FFB900" />
+        </BrandSvg>
+      )
+    case "meta":
+      return (
+        <BrandSvg size={size}>
+          <path d="M4 14.5c1.8-4.2 4-7.5 6.4-7.5 1.6 0 2.5 1.3 4.6 6.3 1.4 3.4 2.2 4.7 3.4 4.7 1.8 0 3.6-2.6 4.6-5" fill="none" stroke="#0081FB" strokeWidth="2.2" strokeLinecap="round" />
+        </BrandSvg>
+      )
+    case "telegram":
+      return (
+        <BrandSvg size={size}>
+          <circle cx="12" cy="12" r="10" fill="#229ED9" />
+          <path d="M7.2 12.1 16.8 8.4 15 16.2l-3.1-1.8-1.6 1.6-.2-2.6Z" fill="#fff" />
+        </BrandSvg>
+      )
+    case "discord":
+      return (
+        <BrandSvg size={size}>
+          <path d="M7.2 5.8 8.6 4.6c2.1.8 4.2 1.2 6.4 1.2h.8L17 5.8c1.8 2.4 2.6 5.4 2.4 8.6-1.6 1.2-3.3 2.1-5.2 2.6L13 15.2c.7-.2 1.3-.6 1.8-1.1-2 .9-4.2.9-6.2 0 .5.5 1.1.9 1.8 1.1L8.8 17c-1.9-.5-3.6-1.4-5.2-2.6C3.4 11.2 4.2 8.2 6 5.8Z" fill="#5865F2" />
+          <circle cx="9.2" cy="11.2" r="1.2" fill="#fff" />
+          <circle cx="14.8" cy="11.2" r="1.2" fill="#fff" />
+        </BrandSvg>
+      )
+    case "twitch":
+      return (
+        <BrandSvg size={size}>
+          <path d="M5 4h14v10.2l-4 4H11l-2.2 2.2H7.2V18.2H5Z" fill="#9146FF" />
+          <path d="M7.4 6.4h1.8v5.2H7.4Zm4 0h1.8v5.2H11.4Z" fill="#fff" />
+        </BrandSvg>
+      )
+    case "tiktok":
+      return (
+        <BrandSvg size={size}>
+          <path d="M14.2 4v9.1a3.3 3.3 0 1 1-2.8-3.3V7.2c1.6.9 3.2 1.4 5 1.5V5.4c-1.4-.1-2.7-.6-3.8-1.4H14.2Z" fill="#25F4EE" />
+          <path d="M13.4 4v9.1a3.3 3.3 0 1 1-2.8-3.3V7.2c1.6.9 3.2 1.4 5 1.5V5.4c-1.4-.1-2.7-.6-3.8-1.4H13.4Z" fill="#FE2C55" transform="translate(1.2 1)" />
+        </BrandSvg>
+      )
+    default:
+      return <GenericCloud size={size} />
+  }
+}
@@ -47,6 +47,8 @@ function NetflowSettingsPanel({
  const [endpoint, setEndpoint] = useState("")
  const [retention, setRetention] = useState("24")
  const [topN, setTopN] = useState("200")
+  const [shareOn, setShareOn] = useState(true)
+  const [sharePct, setSharePct] = useState("5")
  const [ingestOn, setIngestOn] = useState(false)
  const [purgeOpen, setPurgeOpen] = useState(false)
  const [purgeBusy, setPurgeBusy] = useState(false)
@@ -62,6 +64,9 @@ function NetflowSettingsPanel({
    setEndpoint(s.publicEndpoint)
    setRetention(String(s.retentionHours))
    setTopN(String(s.topN))
+    const pct = Number(s.mapServiceMinSharePct ?? 5)
+    setShareOn(pct > 0)
+    setSharePct(String(pct > 0 ? pct : 5))
    setIngestOn(s.enabled)
  }, [backendUrl, enabled])

@@ -83,6 +88,9 @@ function NetflowSettingsPanel({
        publicEndpoint: endpoint,
        retentionHours: Number.parseInt(retention, 10) || 24,
        topN: Number.parseInt(topN, 10) || 200,
+        mapServiceMinSharePct: shareOn
+          ? Math.min(100, Math.max(1, Number.parseFloat(sharePct) || 5))
+          : 0,
      })
      setSettings(res.settings)
      toast.success("Настройки NetFlow сохранены")
@@ -193,6 +201,29 @@ function NetflowSettingsPanel({
          <FormField label="Top-N разговоров">
            <Input value={topN} onChange={(e) => setTopN(e.target.value)} inputMode="numeric" />
          </FormField>
+          <div className="sm:col-span-2 flex flex-col gap-2">
+            <div className="flex items-center gap-3">
+              <FormToggle
+                checked={shareOn}
+                onChange={(on) => {
+                  setShareOn(on)
+                  if (on && (!sharePct || sharePct === "0")) setSharePct("5")
+                }}
+              />
+              <span className="text-sm">Порог доли на карте</span>
+            </div>
+            <FormField
+              label="Минимум % окна"
+              hint="Узел сервиса, если доля байт окна ≥ N%. Выключить — показать все распознанные бренды (макс. 20)"
+            >
+              <Input
+                value={sharePct}
+                onChange={(e) => setSharePct(e.target.value)}
+                inputMode="decimal"
+                disabled={!shareOn}
+              />
+            </FormField>
+          </div>
        </div>

        <p className="text-xs text-muted-foreground">
@@ -0,0 +1,103 @@
+import type { FlowMapHop } from "@mmapp/contracts/traffic-flow"
+import { fmtRate } from "@/lib/fmt-rate"
+
+export interface MatchedNetflowHop {
+  bps: number
+  bpsFwd: number
+  bpsRev: number
+  bytes: number
+}
+
+function ifaceNorm(s: string | undefined): string {
+  return (s ?? "").trim().toLowerCase()
+}
+
+function pairKey(a: string, b: string): string {
+  const x = String(a)
+  const y = String(b)
+  return x <= y ? `${x}\t${y}` : `${y}\t${x}`
+}
+
+function mergeDirected(hops: FlowMapHop[], mapFromId: string): MatchedNetflowHop {
+  let bytes = 0
+  let bpsFwd = 0
+  let bpsRev = 0
+  const from = String(mapFromId)
+  for (const h of hops) {
+    bytes += h.bytes
+    if (h.fromId === from) {
+      bpsFwd += h.bpsFwd
+      bpsRev += h.bpsRev
+    } else {
+      bpsFwd += h.bpsRev
+      bpsRev += h.bpsFwd
+    }
+  }
+  return { bytes, bpsFwd, bpsRev, bps: bpsFwd + bpsRev }
+}
+
+export function hopHasRate(h: MatchedNetflowHop | undefined): h is MatchedNetflowHop {
+  return h != null && Number.isFinite(h.bps) && h.bps > 0
+}
+
+export function formatNetflowRate(hop: MatchedNetflowHop): string {
+  return fmtRate(hop.bps / 1_000_000)
+}
+
+export function formatNetflowDir(hop: MatchedNetflowHop): string {
+  return `↓${fmtRate(hop.bpsFwd / 1_000_000)} ↑${fmtRate(hop.bpsRev / 1_000_000)}`
+}
+
+/** GRE: сначала имя интерфейса туннеля на любом конце, иначе пара узлов. */
+export function matchNetflowForGreEdge(
+  edge: {
+    tunnel: { name: string }
+    fromServer: { id: string }
+    toServer: { id: string }
+  },
+  hops: FlowMapHop[],
+): MatchedNetflowHop | undefined {
+  const name = ifaceNorm(edge.tunnel.name)
+  const fromId = String(edge.fromServer.id)
+  const toId = String(edge.toServer.id)
+  if (name) {
+    const ifaceHits = hops.filter((h) =>
+      h.kind === "iface"
+      && ifaceNorm(h.iface) === name
+      && (h.fromId === fromId || h.fromId === toId),
+    )
+    if (ifaceHits.length) return mergeDirected(ifaceHits, fromId)
+    const greNamed = hops.filter((h) =>
+      h.kind === "gre"
+      && ifaceNorm(h.iface) === name
+      && (h.fromId === fromId || h.fromId === toId || h.toId === fromId || h.toId === toId),
+    )
+    if (greNamed.length) return mergeDirected(greNamed, fromId)
+  }
+  const want = pairKey(fromId, toId)
+  const pairHits = hops.filter((h) =>
+    h.kind === "gre" && Boolean(h.toId) && pairKey(h.fromId, h.toId) === want,
+  )
+  if (pairHits.length) return mergeDirected(pairHits, fromId)
+  return undefined
+}
+
+/** WAN-аплинк HR: kind wan, иначе iface с тем же именем на homeId. */
+export function matchNetflowForWan(
+  homeId: string,
+  wanIface: string,
+  hops: FlowMapHop[],
+): MatchedNetflowHop | undefined {
+  const id = String(homeId)
+  const iface = ifaceNorm(wanIface)
+  if (!iface) return undefined
+  const wanHits = hops.filter((h) =>
+    h.kind === "wan" && h.fromId === id && ifaceNorm(h.iface) === iface,
+  )
+  if (wanHits.length) return mergeDirected(wanHits, id)
+  const ifaceHits = hops.filter((h) =>
+    h.kind === "iface" && h.fromId === id && ifaceNorm(h.iface) === iface,
+  )
+  if (ifaceHits.length) return mergeDirected(ifaceHits, id)
+  return undefined
+}
@@ -35,9 +35,14 @@ export function greTunnelProbe(t: GreTunnel): TunnelProbe {
  }
 }

-const W = 1060
+const W = 1240
 const H = 580
 const MARGIN = 72
+const SERVICE_COL_W = 150
+
+/** Карточка конечного сервиса на карте (центр = позиция узла). */
+export const MAP_SERVICE_NODE_W = 86
+export const MAP_SERVICE_NODE_H = 58

 /** Одна горизонтальная «полка» на карте: Home → JH → Exit слева направо. */
 export const NETWORK_MAP_PIPELINE_Y = 300
@@ -68,7 +73,9 @@ function layerOfServer(s: Server): number | null {
 * Увеличивать при изменении алгоритма раскладки спутников/узлов.
 * Страница карты сбрасывает сохранённые перетаскивания при смене значения (в т.ч. после hot reload).
 */
-export const NETWORK_MAP_LAYOUT_REVISION = 6
+export const NETWORK_MAP_W = W
+export const NETWORK_MAP_H = H
+export const NETWORK_MAP_LAYOUT_REVISION = 7

 export interface WanJhEdge {
  homeId: string
@@ -254,7 +261,7 @@ export function computeNetworkMapLayout(
  const nodePos: Record<string, { x: number; y: number }> = {}
  const wanSatPos: Record<string, { x: number; y: number }[]> = {}

-  const span = W - 2 * MARGIN
+  const span = W - 2 * MARGIN - SERVICE_COL_W
  const laneGap = Math.min(44, span * 0.04)
  const laneW = (span - 2 * laneGap) / 3

@@ -425,6 +432,28 @@ export function computeNetworkMapLayout(
  return { nodePos, wanSatPos }
 }

+/** Колонка конечных сервисов справа от EN. */
+export function placeServiceNodes(
+  serviceIds: string[],
+  enPositions: Array<{ x: number; y: number }>,
+): Record<string, { x: number; y: number }> {
+  const out: Record<string, { x: number; y: number }> = {}
+  if (serviceIds.length === 0) return out
+  const minY = MARGIN + 70
+  const maxY = H - 72
+  const x = W - MARGIN - SERVICE_COL_W / 2
+  const enYs = enPositions.map((p) => p.y).filter((y) => Number.isFinite(y))
+  const centerY = enYs.length ? enYs.reduce((a, b) => a + b, 0) / enYs.length : (minY + maxY) / 2
+  const n = serviceIds.length
+  const gap = Math.min(96, (maxY - minY) / Math.max(1, n))
+  const span = gap * (n - 1)
+  const start = clamp(centerY - span / 2, minY, maxY - span)
+  serviceIds.forEach((id, i) => {
+    out[id] = { x, y: n === 1 ? clamp(centerY, minY, maxY) : start + i * gap }
+  })
+  return out
+}
+
 /**
 * Суммарная задержка «дом → JH» в миллисекундах: те же поля `Server.latency`, что показываются в разделе Серверы.
 * Отдельного ICMP по ребру нет — это не замер линии, а сумма каталожных latency концов.
@@ -803,3 +832,42 @@ export function buildGreMapEdges(
  }
  return out
 }
+
+/**
+ * Обрезать отрезок центр круга → центр прямоугольника по ободу круга и AABB карточки.
+ * Пунктир EN→сервис визуально упирается в край, как GRE под кругами узлов.
+ */
+export function clipSegmentCircleToRect(
+  x1: number,
+  y1: number,
+  r: number,
+  x2: number,
+  y2: number,
+  hw: number,
+  hh: number,
+  pad = 1.5,
+): { x1: number; y1: number; x2: number; y2: number } {
+  const dx = x2 - x1
+  const dy = y2 - y1
+  const len = Math.hypot(dx, dy)
+  if (len < 1e-6) return { x1, y1, x2, y2 }
+  const ux = dx / len
+  const uy = dy / len
+  const sx = x1 + ux * (r + pad)
+  const sy = y1 + uy * (r + pad)
+  const absDx = Math.abs(dx)
+  const absDy = Math.abs(dy)
+  const u = Math.min(
+    absDx < 1e-9 ? 1 : (hw + pad) / absDx,
+    absDy < 1e-9 ? 1 : (hh + pad) / absDy,
+  )
+  const uu = Math.min(Math.max(u, 0), 0.48)
+  const ex = x2 - dx * uu
+  const ey = y2 - dy * uu
+  if ((ex - sx) * dx + (ey - sy) * dy <= 0) {
+    const mx = (x1 + x2) / 2
+    const my = (y1 + y2) / 2
+    return { x1: mx - ux * 2, y1: my - uy * 2, x2: mx + ux * 2, y2: my + uy * 2 }
+  }
+  return { x1: sx, y1: sy, x2: ex, y2: ey }
+}
@@ -14259,6 +14259,21 @@
      "dependencies": {
        "zod": "^4.4.1"
      }
+    },
+    "node_modules/@next/swc-win32-x64-msvc": {
+      "version": "16.2.4",
+      "resolved": "https://registry.npmjs.org/@next/swc-win32-x64-msvc/-/swc-win32-x64-msvc-16.2.4.tgz",
+      "integrity": "sha512-kMVGgsqhO5YTYODD9IPGGhA6iprWidQckK3LmPeW08PIFENRmgfb4MjXHO+p//d+ts2rpjvK5gXWzXSMrPl9cw==",
+      "cpu": [
+        "x64"
+      ],
+      "optional": true,
+      "os": [
+        "win32"
+      ],
+      "engines": {
+        "node": ">= 10"
+      }
    }
  }
 }
@@ -21,6 +21,7 @@ export const trafficFlowSettingsDtoSchema = z.object({
  hubServerId: z.number().int().positive().nullable(),
  retentionHours: z.number().int().positive(),
  topN: z.number().int().positive(),
+  mapServiceMinSharePct: z.number().min(0).max(100),
  lastDatagramAt: z.string().nullable(),
  lastExporterIp: z.string().nullable(),
  lastError: z.string().nullable(),
@@ -40,6 +41,7 @@ export const trafficFlowSettingsPatchSchema = z.object({
  hubServerId: z.number().int().positive().nullable().optional(),
  retentionHours: z.number().int().positive().optional(),
  topN: z.number().int().positive().max(1000).optional(),
+  mapServiceMinSharePct: z.number().min(0).max(100).optional(),
 })

 export const trafficFlowOverlayRequestSchema = z.object({
@@ -248,6 +250,72 @@ export const flowPurgeDtoSchema = z.object({
  vacuumed: z.boolean(),
 })

+export const flowMapHopKindSchema = z.enum(["gre", "wan", "iface"])
+
+export const flowMapHopDtoSchema = z.object({
+  fromId: z.string(),
+  fromLabel: z.string(),
+  toId: z.string(),
+  toLabel: z.string(),
+  kind: flowMapHopKindSchema,
+  iface: z.string().optional(),
+  bytes: z.number().nonnegative(),
+  bps: z.number().nonnegative(),
+  bpsFwd: z.number().nonnegative(),
+  bpsRev: z.number().nonnegative(),
+})
+
+export const flowMapServiceDtoSchema = z.object({
+  id: z.string(),
+  label: z.string(),
+  category: z.string(),
+  bytes: z.number().nonnegative(),
+  bps: z.number().nonnegative(),
+  share: z.number().min(0).max(1),
+})
+
+export const flowMapServiceEdgeDtoSchema = z.object({
+  fromId: z.string(),
+  toId: z.string(),
+  bytes: z.number().nonnegative(),
+  bps: z.number().nonnegative(),
+  bpsFwd: z.number().nonnegative(),
+  bpsRev: z.number().nonnegative(),
+  clientId: z.string().optional(),
+  clientName: z.string().optional(),
+  clients: z.array(z.object({
+    id: z.string(),
+    name: z.string(),
+  })).optional(),
+})
+
+export const flowMapServicePathDtoSchema = z.object({
+  clientId: z.string(),
+  clientName: z.string(),
+  viaId: z.string(),
+  viaName: z.string(),
+  enId: z.string(),
+  enName: z.string(),
+  serviceId: z.string(),
+  bytes: z.number().nonnegative(),
+  bps: z.number().nonnegative(),
+})
+
+export const flowMapHopsDtoSchema = z.object({
+  hops: z.array(flowMapHopDtoSchema),
+  live: z.boolean(),
+  rangeMinutes: z.number().int().positive(),
+  windowSec: z.number().positive(),
+  totalBytes: z.number().nonnegative().optional(),
+  services: z.array(flowMapServiceDtoSchema).optional(),
+  serviceEdges: z.array(flowMapServiceEdgeDtoSchema).optional(),
+  servicePaths: z.array(flowMapServicePathDtoSchema).optional(),
+  mapServiceMinSharePct: z.number().min(0).max(100).optional(),
+  dedupApplied: z.boolean(),
+  excludeMeshApplied: z.boolean(),
+  excludeOverlayApplied: z.boolean(),
+})
+
 export type FlowTalkerDto = z.infer<typeof flowTalkerDtoSchema>
 export type FlowStatsDto = z.infer<typeof flowStatsDtoSchema>
 export type FlowBreakdownRow = z.infer<typeof flowBreakdownRowSchema>
@@ -260,3 +328,9 @@ export type FlowExportersDto = z.infer<typeof flowExportersDtoSchema>
 export type FlowClientsDto = z.infer<typeof flowClientsDtoSchema>
 export type FlowMonthlyDto = z.infer<typeof flowMonthlyDtoSchema>
 export type FlowPurgeDto = z.infer<typeof flowPurgeDtoSchema>
+export type FlowMapHopKind = z.infer<typeof flowMapHopKindSchema>
+export type FlowMapHop = z.infer<typeof flowMapHopDtoSchema>
+export type FlowMapService = z.infer<typeof flowMapServiceDtoSchema>
+export type FlowMapServiceEdge = z.infer<typeof flowMapServiceEdgeDtoSchema>
+export type FlowMapServicePath = z.infer<typeof flowMapServicePathDtoSchema>
+export type FlowMapHopsDto = z.infer<typeof flowMapHopsDtoSchema>
@@ -2,6 +2,7 @@ import type {
  FlowAnalyticsDto,
  FlowClientsDto,
  FlowExportersDto,
+  FlowMapHopsDto,
  FlowMonthlyDto,
  FlowPurgeDto,
  FlowStatsDto,
@@ -88,6 +89,29 @@ export async function getFlowClients(baseUrl: string, range = "5m"): Promise<Flo
  return requestJson<FlowClientsDto>(baseUrl, `/api/traffic/flow/clients?range=${encodeURIComponent(range)}`)
 }

+export async function getFlowMapHops(
+  baseUrl: string,
+  params: {
+    range?: string
+    serverId?: string
+    userId?: string
+    iface?: string
+    dedup?: boolean
+    excludeMesh?: boolean
+    excludeOverlay?: boolean
+  } = {},
+): Promise<FlowMapHopsDto> {
+  return requestJson<FlowMapHopsDto>(baseUrl, `/api/traffic/flow/map-hops${flowQuery({
+    range: params.range ?? "5m",
+    serverId: params.serverId,
+    userId: params.userId,
+    iface: params.iface,
+    dedup: params.dedup,
+    excludeMesh: params.excludeMesh,
+    excludeOverlay: params.excludeOverlay,
+  })}`)
+}
+
 export async function getFlowAnalytics(
  baseUrl: string,
  params: {
Author	SHA1	Message	Date
DenozordecandCursor	0e5fb065e2	fix(backend): уменьшить лишние записи SQLite Docker images / prepare-release (push) Successful in 10s Details Docker images / backend-image (push) Successful in 2m8s Details Docker images / frontend-image (push) Successful in 3m14s Details Docker images / notify-webhook (push) Skipped Details Docker images / updater-image (push) Successful in 47s Details Docker images / publish-release (push) Successful in 12s Details Убрать WAL TRUNCATE с hot path NetFlow, не писать неизменённые UPDATE и служебные INSERT, кэшировать карты topology/analytics. Co-authored-by: Cursor <[email protected]>	2026-09-07 20:22:20 +07:00
DenozordecandCursor	5188b2aff2	fix(network-map): обрезать пунктир до сервиса и показать пути клиентов Docker images / prepare-release (push) Successful in 8s Details Docker images / backend-image (push) Successful in 1m58s Details Docker images / frontend-image (push) Successful in 3m17s Details Docker images / notify-webhook (push) Skipped Details Docker images / updater-image (push) Successful in 49s Details Docker images / publish-release (push) Successful in 13s Details Линия от обода EN до рамки сервиса; таблица клиент, узел и сервис с подсветкой на карте. Co-authored-by: Cursor <[email protected]>	2026-09-07 17:39:18 +07:00
DenozordecandCursor	cd1fd2c9d3	fix(network-map): рисовать пунктир и скорость до сервиса Docker images / prepare-release (push) Successful in 8s Details Docker images / backend-image (push) Successful in 1m41s Details Docker images / frontend-image (push) Successful in 2m55s Details Docker images / notify-webhook (push) Skipped Details Docker images / updater-image (push) Successful in 43s Details Docker images / publish-release (push) Successful in 9s Details Якорь hop на EN даже без nextHop; линия ARN\to Google с NetflowRateBadge. Co-authored-by: Cursor <[email protected]>	2026-09-07 17:02:28 +07:00
DenozordecandCursor	77425cca32	fix(network-map): якорить сервисы на выходную ноду Docker images / prepare-release (push) Successful in 8s Details Docker images / backend-image (push) Successful in 1m46s Details Docker images / frontend-image (push) Successful in 2m56s Details Docker images / notify-webhook (push) Skipped Details Docker images / updater-image (push) Successful in 45s Details Docker images / publish-release (push) Successful in 16s Details Пунктир от EN, а не от JH; перетаскивание узлов сервисов; иконка Google без foreignObject. Co-authored-by: Cursor <[email protected]>	2026-09-07 16:10:39 +07:00
DenozordecandCursor	29d245cde3	feat(traffic): enhance flow analytics and brand classification Docker images / prepare-release (push) Successful in 11s Details Docker images / backend-image (push) Successful in 2m12s Details Docker images / frontend-image (push) Successful in 4m3s Details Docker images / notify-webhook (push) Skipped Details Docker images / updater-image (push) Successful in 48s Details Docker images / publish-release (push) Successful in 12s Details - Introduced `pickInternetPeer` function to improve IP classification by selecting the appropriate public IP from source and destination. - Updated `buildFlowAnalytics` and related functions to utilize the new peer selection logic, enhancing accuracy in flow analytics. - Added tests for new classifications and ensured existing tests cover new scenarios for Google and Cloudflare. - Refactored traffic flow brand mappings to include additional CIDR ranges for Google and Cloudflare. Co-authored-by: Cursor <[email protected]>	2026-09-07 15:35:52 +07:00
DenozordecandCursor	7a491a325d	fix(network-map): убрать тяжёлую классификацию из poll карты Docker images / prepare-release (push) Successful in 9s Details Docker images / backend-image (push) Successful in 1m53s Details Docker images / frontend-image (push) Successful in 3m11s Details Docker images / notify-webhook (push) Skipped Details Docker images / updater-image (push) Successful in 49s Details Docker images / publish-release (push) Successful in 12s Details Не сканировать RIPE и каталог по каждой строке окна; сервисы по уникальным dst и кэшу ASN. Порог доли на карте настраиваемый и отключаемый. Co-authored-by: Cursor <[email protected]>	2026-09-07 14:17:04 +07:00
DenozordecandCursor	db64621122	feat(network-map): показать конечные сервисы на карте сети Docker images / prepare-release (push) Successful in 9s Details Docker images / backend-image (push) Successful in 2m0s Details Docker images / frontend-image (push) Successful in 3m15s Details Docker images / notify-webhook (push) Skipped Details Docker images / updater-image (push) Successful in 45s Details Docker images / publish-release (push) Successful in 12s Details NetFlow ≥ 5% окна, узлы с логотипом бренда справа от EN, скорость потока на рёбрах к сервисам. Co-authored-by: Cursor <[email protected]>	2026-09-07 13:30:45 +07:00
DenozordecandCursor	6332d83a12	feat(traffic): показать поток NetFlow на карте сети Docker images / prepare-release (push) Successful in 10s Details Docker images / backend-image (push) Successful in 2m14s Details Docker images / frontend-image (push) Successful in 3m14s Details Docker images / notify-webhook (push) Skipped Details Docker images / updater-image (push) Successful in 46s Details Docker images / publish-release (push) Successful in 11s Details Скорость между узлами считается как в Трафике (5 мин, без overlay/mesh), отдельно от ёмкости WAN и BT. Co-authored-by: Cursor <[email protected]>	2026-09-07 12:37:47 +07:00