feat(netflow): автоматизировать настройку туннеля коллектора

backend/.env.example

@@ -17,3 +17,5 @@ AUTH_REQUIRED=false
 AUTH_JWT_SECRET=dev-secret-change-me
 AUTH_ISSUER=https://auth.shnt.top
 AUTH_PORTAL_URL=http://localhost:5175
+# Optional: enabled by the NetFlow agent Compose installer. Linux Unix socket only.
+# FLOW_AGENT_SOCKET=/run/mm-netflow-host/agent.sock

backend/Dockerfile

@@ -57,6 +57,7 @@ COPY --from=build /app/packages/contracts ./packages/contracts
 COPY --from=build /app/backend/dist ./backend/dist
 COPY --from=build /app/backend/package.json ./backend/package.json
 COPY --from=build /app/backend/drizzle ./backend/drizzle
+COPY deploy/netflow-agent/install.sh ./deploy/netflow-agent/install.sh
 # Nested install from lockfile (dotenv etc.) — ESM resolves from /app/backend/dist → ../node_modules
 COPY --from=build /app/backend/node_modules ./backend/node_modules
 EXPOSE 8000

backend/package.json

@@ -33,7 +33,7 @@
     "test": "node scripts/test.mjs",
     "test:evolution": "tsx src/services/evobgp-catalog.test.ts && tsx src/routes/evobgp-catalog.test.ts && tsx src/routes/evobgp-settings.test.ts && tsx src/plugins/auth.evolution-settings.test.ts",
     "test:netflow-purge": "tsx src/db/operation-lock.test.ts && tsx src/services/traffic-flow-purge-execution.test.ts && tsx src/routes/traffic-flow-purge.test.ts",
-    "test:netflow-settings": "tsx src/services/traffic-flow-host-file-builders.test.ts && tsx src/services/traffic-flow-key-execution.test.ts && tsx src/routes/traffic-flow-settings.test.ts && tsx src/plugins/auth.netflow-settings.test.ts",
+    "test:netflow-settings": "tsx src/services/traffic-flow-host-file-builders.test.ts && tsx src/services/traffic-flow-key-execution.test.ts && tsx src/routes/traffic-flow-settings.test.ts && tsx src/plugins/auth.netflow-settings.test.ts && tsx src/services/traffic-flow-agent-install.test.ts && tsx src/services/traffic-flow-agent-client.test.ts && tsx src/services/traffic-flow-operation-queue.test.ts && tsx src/routes/traffic-flow-host.test.ts",
     "test:geoip": "tsx src/services/geoip-download.test.ts && tsx src/services/traffic-flow-geoip.test.ts && tsx src/plugins/auth.geoip.test.ts",
     "test:ospf-optimizer": "tsx src/services/ospf-optimizer-identity.test.ts && tsx src/services/ospf-optimizer-metrics.test.ts && tsx src/services/ospf-optimizer-plan.test.ts && tsx src/services/ospf-effective-disabled.test.ts && tsx src/routes/ospf-optimizer.test.ts",
     "test:containers": "tsx src/services/containers-live.test.ts && tsx src/services/containers-state.test.ts && tsx src/services/container-removal.test.ts && tsx src/routes/containers-state.test.ts",

backend/src/index.ts

@@ -39,6 +39,8 @@ import usersRoutes from "./routes/users.js"
 import statisticsRoutes from "./routes/statistics.js"
 import { refreshScheduler, stopScheduler } from "./services/scheduler.js"
 import { getFlowWorkerHealth, startTrafficFlowListener, stopTrafficFlowListener } from "./services/traffic-flow-ingest.js"
+import { agentSocket } from "./services/traffic-flow-agent-client.js"
+import { reconcileFlowHost, withFlowConfigurationLock } from "./services/traffic-flow-host-agent.js"
 import { initGeoip } from "./services/traffic-flow-geoip.js"
 
 const eventLoopDelay = monitorEventLoopDelay({ resolution: 20 })
@@ -147,6 +149,10 @@ export async function buildApp(opts?: {
     await refreshScheduler()
     await initGeoip()
     await startTrafficFlowListener()
+    if (agentSocket()) {
+      try { await withFlowConfigurationLock(reconcileFlowHost) }
+      catch { app.log.warn("NetFlow: агент хоста не синхронизирован; проверьте состояние в настройках NetFlow") }
+    }
     app.addHook("onClose", async () => {
       stopScheduler()
       stopTrafficFlowListener()

backend/src/routes/traffic-flow-host.test.ts

@@ -0,0 +1,46 @@
+import assert from "node:assert/strict"
+import Fastify from "fastify"
+import { SignJWT } from "jose"
+import { createTrafficFlowHostRoutes } from "./traffic-flow-host.js"
+import { emptyHostStatus } from "../services/traffic-flow-agent-client.js"
+
+process.env.AUTH_REQUIRED = "true"
+process.env.AUTH_JWT_SECRET = "flow-host-test-secret"
+process.env.AUTH_ISSUER = "https://auth.test.local"
+process.env.NODE_ENV = "test"
+const { default: authPlugin, requireAuth } = await import("../plugins/auth.js")
+const app = Fastify()
+await app.register(authPlugin)
+app.addHook("preHandler", requireAuth)
+let applied = 0, fail = false
+let status = emptyHostStatus(false)
+await app.register(createTrafficFlowHostRoutes({
+  status: async () => status,
+  reconcile: async () => { applied++; if (fail) throw Error("Туннель не изменён: конфликт адресов"); return status },
+}), { prefix: "/api" })
+async function token(permission: string) {
+  return new SignJWT({ sub: "fixture", apps: ["mm"], permissions: [permission] }).setProtectedHeader({ alg: "HS256" })
+    .setIssuer("https://auth.test.local").setExpirationTime("1h").sign(new TextEncoder().encode("flow-host-test-secret"))
+}
+for (const permission of ["mm:dashboard:read", "mm:traffic:read", "mm:traffic:write", "mm:traffic:admin"]) {
+  const headers = { authorization: "Bearer " + await token(permission) }
+  assert.equal((await app.inject({ method: "GET", url: "/api/traffic/flow/host-status", headers })).statusCode, permission === "mm:dashboard:read" ? 403 : 200)
+  const before = applied
+  const response = await app.inject({ method: "POST", url: "/api/traffic/flow/host-reconcile", headers })
+  const allowed = permission.endsWith(":write") || permission.endsWith(":admin")
+  assert.equal(response.statusCode, allowed ? 409 : 403)
+  assert.equal(applied, before + Number(allowed))
+}
+assert.equal((await app.inject({ method: "GET", url: "/api/traffic/flow/host-status" })).statusCode, 401)
+const headers = { authorization: "Bearer " + await token("mm:traffic:write") }
+status = { ...emptyHostStatus(true), reachable: true, interfaceUp: true, configMatches: true }
+let response = await app.inject({ method: "POST", url: "/api/traffic/flow/host-reconcile", headers })
+assert.equal(response.statusCode, 200); assert.equal(response.json().hostStatus.configMatches, true)
+fail = true
+response = await app.inject({ method: "POST", url: "/api/traffic/flow/host-reconcile", headers })
+assert.equal(response.statusCode, 502); assert.match(response.json().error, /конфликт адресов/)
+status = emptyHostStatus(true, "offline")
+response = await app.inject({ method: "GET", url: "/api/traffic/flow/host-status", headers })
+assert.equal(response.json().configMatches, null); assert.equal(response.json().reachable, false)
+await app.close()
+console.log("NetFlow host HTTP/JWT: read/write permissions, manual mode, reconcile success/conflict and offline status")

backend/src/routes/traffic-flow-host.ts

@@ -0,0 +1,20 @@
+import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
+import type { TrafficFlowHostStatus } from "@mmapp/contracts/traffic-flow"
+
+export function createTrafficFlowHostRoutes(deps: {
+  status: () => Promise<TrafficFlowHostStatus>
+  reconcile: () => Promise<TrafficFlowHostStatus>
+}): FastifyPluginAsyncZod {
+  return async app => {
+    app.get("/traffic/flow/host-status", async () => deps.status())
+    app.post("/traffic/flow/host-reconcile", async (_req, reply) => {
+      try {
+        const hostStatus = await deps.reconcile()
+        if (!hostStatus.configured) return reply.status(409).send({ error: "Автоматизация хоста NetFlow не настроена" })
+        return { ok: true, hostStatus }
+      } catch (error) {
+        return reply.status(502).send({ error: error instanceof Error ? error.message : "Настройки хоста не применены" })
+      }
+    })
+  }
+}

backend/src/routes/traffic-flow.ts

@@ -8,6 +8,9 @@ import {
 } from "@mmapp/contracts/traffic-flow"
 import { createTrafficFlowPurgeRoutes } from "./traffic-flow-purge.js"
 import { createTrafficFlowSettingsRoutes } from "./traffic-flow-settings.js"
+import { createTrafficFlowHostRoutes } from "./traffic-flow-host.js"
+import { getFlowHostStatus, reconcileFlowHost, withFlowConfigurationLock } from "../services/traffic-flow-host-agent.js"
+import { agentSocket } from "../services/traffic-flow-agent-client.js"
 import {
   ensureHostKeys,
   getTrafficFlowSettingsRow,
@@ -150,11 +153,21 @@ function sleep(ms: number, signal: AbortSignal): Promise<void> {
 }
 
 const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => {
+  await app.register(createTrafficFlowHostRoutes({
+    status: getFlowHostStatus,
+    reconcile: () => withFlowConfigurationLock(reconcileFlowHost),
+  }))
   await app.register(createTrafficFlowSettingsRoutes({
     read: () => toTrafficFlowSettingsDto(getFlowListenerState()),
-    update: async patch => { await updateTrafficFlowSettings(patch) },
+    update: async patch => { await withFlowConfigurationLock(async () => {
+      const current = await getTrafficFlowSettingsRow()
+      if (agentSocket() && (["collectorIp", "prefix", "flowListenPort", "wgListenPort"] as const).some(key => patch[key] !== undefined && patch[key] !== current[key])) {
+        throw Object.assign(new Error("Адрес, префикс и порты закреплены установкой агента; повторите установщик хоста"), { statusCode: 409 })
+      }
+      await updateTrafficFlowSettings(patch)
+    }) },
     startListener: startTrafficFlowListener,
-    ensureKeys: ensureHostKeys,
+    ensureKeys: () => withFlowConfigurationLock(ensureHostKeys),
     hasPrivateKey: async () => Boolean((await getTrafficFlowSettingsRow()).hostPrivateKey),
     hostFiles: listTrafficFlowHostFiles,
   }))

backend/src/services/traffic-flow-agent-client.test.ts

@@ -0,0 +1,42 @@
+import assert from "node:assert/strict"
+import { createServer } from "node:http"
+import { randomUUID } from "node:crypto"
+import { tmpdir } from "node:os"
+import { join } from "node:path"
+import { callFlowAgent, emptyHostStatus, type FlowAgentSnapshot } from "./traffic-flow-agent-client.js"
+
+const socket = process.platform === "win32" ? `\\\\.\\pipe\\flow-agent-${randomUUID()}` : join(tmpdir(), `flow-agent-${randomUUID()}.sock`)
+const snapshot: FlowAgentSnapshot = {
+  collectorIp: "10.255.254.1", prefix: "10.255.254.0/24", flowListenPort: 4739, wgListenPort: 51821,
+  hostPublicKey: "public-fixture", hostPrivateKey: "private-fixture", peers: [],
+}
+let mode = "ok"
+const server = createServer((req, res) => {
+  let body = ""
+  req.on("data", chunk => { body += chunk })
+  req.on("end", () => {
+    assert.equal(req.url, "/check"); assert.equal(req.method, "POST")
+    assert.deepEqual(JSON.parse(body), snapshot)
+    if (mode === "conflict") { res.writeHead(409); res.end(JSON.stringify({ error: "Конфликт адресов" })); return }
+    if (mode === "invalid") { res.end("not-json"); return }
+    if (mode === "oversize") { res.end("x".repeat(300_000)); return }
+    res.end(JSON.stringify({
+      interfaceUp: true, configMatches: true, publicKey: "public-fixture", listenPort: 51821,
+      addresses: ["10.255.254.1/24"], peers: [{ publicKey: "router-fixture", allowedIps: ["10.255.254.2/32"], lastHandshakeAt: null, receivedBytes: 0, sentBytes: 0 }],
+      privateKey: "must-not-be-exposed",
+    }))
+  })
+})
+await new Promise<void>((resolve, reject) => { server.once("error", reject); server.listen(socket, resolve) })
+try {
+  const status = await callFlowAgent("/check", snapshot, socket)
+  assert.equal(status.reachable, true); assert.equal(status.configMatches, true)
+  assert.equal(JSON.stringify(status).includes("must-not-be-exposed"), false)
+  assert.equal(status.peers[0].lastHandshakeAt, null)
+  mode = "conflict"; await assert.rejects(callFlowAgent("/check", snapshot, socket), /Конфликт адресов/)
+  mode = "invalid"; await assert.rejects(callFlowAgent("/check", snapshot, socket), /некорректный ответ/)
+  mode = "oversize"; await assert.rejects(callFlowAgent("/check", snapshot, socket), /некорректный ответ/)
+  await assert.rejects(callFlowAgent("/status", undefined, ""), /не настроена/)
+  assert.equal(emptyHostStatus(true, "offline").configMatches, null)
+} finally { await new Promise<void>(resolve => server.close(() => resolve())) }
+console.log("NetFlow agent IPC: request, validated status, secret stripping, conflict, invalid/oversize response, manual mode")

backend/src/services/traffic-flow-agent-client.ts

@@ -0,0 +1,45 @@
+import { request } from "node:http"
+import { trafficFlowHostStatusSchema, type TrafficFlowHostStatus } from "@mmapp/contracts/traffic-flow"
+
+export type FlowAgentSnapshot = {
+  collectorIp: string; prefix: string; flowListenPort: number; wgListenPort: number
+  hostPublicKey: string; hostPrivateKey: string
+  peers: { serverId: number; publicKey: string; allowedIps: string[]; endpoint?: string }[]
+}
+
+export const agentSocket = () => process.env.FLOW_AGENT_SOCKET?.trim() || ""
+export const emptyHostStatus = (configured: boolean, error: string | null = null): TrafficFlowHostStatus => ({
+  configured, reachable: false, interfaceUp: false, configMatches: null,
+  publicKey: "", listenPort: 0, addresses: [], peers: [], error,
+})
+
+export function callFlowAgent(path: "/status" | "/check" | "/apply", snapshot?: FlowAgentSnapshot, socketPath = agentSocket()): Promise<TrafficFlowHostStatus> {
+  if (!socketPath) return Promise.reject(new Error("Автоматизация хоста NetFlow не настроена"))
+  const body = snapshot ? JSON.stringify(snapshot) : undefined
+  return new Promise((resolve, reject) => {
+    const fail = () => reject(new Error("Агент NetFlow недоступен или вернул некорректный ответ; проверьте контейнер mmapp-netflow-agent"))
+    const req = request({ socketPath, path, method: body ? "POST" : "GET", headers: body ? { "content-type": "application/json", "content-length": Buffer.byteLength(body) } : {} }, res => {
+      let text = ""
+      res.setEncoding("utf8")
+      res.on("data", (chunk: string) => { text += chunk; if (text.length > 262144) req.destroy() })
+      res.on("error", fail)
+      res.on("end", () => {
+        try {
+          const raw: unknown = JSON.parse(text)
+          if (res.statusCode !== 200) {
+            // The local controller deliberately emits sanitized errors only.
+            const error = raw as { error?: unknown }
+            reject(new Error(typeof error.error === "string" ? error.error.slice(0, 500) : "Агент NetFlow отклонил настройки"))
+            return
+          }
+          resolve(trafficFlowHostStatusSchema.parse({ ...(raw as object), configured: true, reachable: true, error: null }))
+        } catch { fail() }
+      })
+    })
+    // An absolute deadline also covers a peer that keeps sending small chunks.
+    const timer = setTimeout(() => req.destroy(), path === "/apply" ? 90_000 : 30_000)
+    req.on("close", () => clearTimeout(timer))
+    req.on("error", fail)
+    req.end(body)
+  })
+}

backend/src/services/traffic-flow-agent-install.test.ts

@@ -0,0 +1,18 @@
+import assert from "node:assert/strict"
+import { readFile } from "node:fs/promises"
+import { buildFlowAgentInstaller } from "./traffic-flow-agent-install.js"
+import { buildTrafficFlowHostFiles } from "./traffic-flow-host-file-builders.js"
+const files = buildTrafficFlowHostFiles({ collectorIp: "10.255.254.1", prefix: "10.255.254.0/24", flowListenPort: 4740,
+  wgListenPort: 51821, hostPrivateKey: Buffer.alloc(32, 1).toString("base64"), peers: [] })
+const conf = files.find(file => file.id === "wg-quick")!.code
+const source = await readFile(new URL("../../../deploy/netflow-agent/install.sh", import.meta.url), "utf8")
+const script = buildFlowAgentInstaller(conf, 4740, source)
+assert.ok(script.startsWith("#!/usr/bin/env bash\n"))
+assert.ok(script.includes(conf.trim() + "\nMM_CONFIG_EOF\n"))
+assert.ok(script.includes('"${1:-/opt/cdn-mm}" 4740'))
+assert.ok(script.includes("wg-quick@wg-flow.service"))
+assert.ok(script.includes("docker-compose.netflow-agent.yml"))
+assert.ok(!script.includes("\r"))
+assert.throws(() => buildFlowAgentInstaller("MM_CONFIG_EOF\n", 4739, source))
+assert.throws(() => buildFlowAgentInstaller(conf, 0, source))
+console.log("NetFlow agent installer export: validated snapshot, custom port/path, embedded installer and LF-only output")

backend/src/services/traffic-flow-agent-install.ts

@@ -0,0 +1,8 @@
+/** Call only with a configuration already validated by buildTrafficFlowHostFiles. */
+export function buildFlowAgentInstaller(conf: string, flowPort: number, installer: string): string {
+  if (!Number.isInteger(flowPort) || flowPort < 1 || flowPort > 65535 || conf.split(/\r?\n/).includes("MM_CONFIG_EOF")) {
+    throw new Error("Некорректная конфигурация установщика NetFlow")
+  }
+  const clean = (text: string) => text.replaceAll("\r\n", "\n").replace(/\s*$/, "") + "\n"
+  return "#!/usr/bin/env bash\nset -euo pipefail\numask 077\nEXPORT_DIR=$(mktemp -d)\ntrap 'rm -rf -- \"$EXPORT_DIR\"' EXIT\ncat > \"$EXPORT_DIR/wg-flow.conf\" <<'MM_CONFIG_EOF'\n" + clean(conf) + "MM_CONFIG_EOF\ncat > \"$EXPORT_DIR/install.sh\" <<'MM_INSTALL_EOF'\n" + clean(installer) + "MM_INSTALL_EOF\nbash \"$EXPORT_DIR/install.sh\" \"$EXPORT_DIR/wg-flow.conf\" \"${1:-/opt/cdn-mm}\" " + flowPort + "\n"
+}

backend/src/services/traffic-flow-host-agent.ts

@@ -0,0 +1,46 @@
+import { withAdvisoryLock } from "../db/index.js"
+import { parseJsonArray } from "../db/json.js"
+import { getTrafficFlowSettingsRow, invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js"
+import { agentSocket, callFlowAgent, emptyHostStatus, type FlowAgentSnapshot } from "./traffic-flow-agent-client.js"
+import { FlowConfigurationQueue } from "./traffic-flow-operation-queue.js"
+
+// Session lock: RouterOS/network operations use other pooled connections. Never lock the settings row here.
+const queue = new FlowConfigurationQueue()
+export const withFlowConfigurationLock = <T>(fn: () => Promise<T>) => queue.run(() => withAdvisoryLock(7340291, async () => {
+  invalidateTrafficFlowSettingsCache()
+  return fn()
+}))
+
+async function snapshot(): Promise<FlowAgentSnapshot> {
+  const row = await getTrafficFlowSettingsRow()
+  return {
+    collectorIp: row.collectorIp, prefix: row.prefix, flowListenPort: row.flowListenPort, wgListenPort: row.wgListenPort,
+    hostPublicKey: row.hostPublicKey, hostPrivateKey: row.hostPrivateKey,
+    peers: (parseJsonArray(row.peersJson) as FlowAgentSnapshot["peers"]).map(p => ({
+      serverId: p.serverId, publicKey: p.publicKey, allowedIps: p.allowedIps, ...(p.endpoint ? { endpoint: p.endpoint } : {}),
+    })),
+  }
+}
+
+export async function getFlowHostStatus() {
+  if (!agentSocket()) return emptyHostStatus(false)
+  try { return await callFlowAgent("/check", await snapshot()) }
+  catch (error) {
+    const message = error instanceof Error ? error.message : "Проверка агента не выполнена"
+    // Preserve live measurements on a settings conflict; never claim configuration agreement.
+    try { return { ...await callFlowAgent("/status"), configMatches: false, error: message } }
+    catch { return emptyHostStatus(true, message) }
+  }
+}
+
+export async function preflightFlowHost() {
+  if (!agentSocket()) return
+  try { await callFlowAgent("/check", await snapshot()) }
+  catch (error) { throw Object.assign(new Error(error instanceof Error ? error.message : "Агент NetFlow не готов"), { statusCode: 409 }) }
+}
+
+// Caller holds withFlowConfigurationLock. This is also used after saving a newly discovered router key.
+export async function reconcileFlowHost() {
+  if (!agentSocket()) return emptyHostStatus(false)
+  return callFlowAgent("/apply", await snapshot())
+}

backend/src/services/traffic-flow-host-files.ts

@@ -1,8 +1,15 @@
 import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js"
 import { parseJsonArray } from "../db/json.js"
 import { buildTrafficFlowHostFiles } from "./traffic-flow-host-file-builders.js"
+import { readFile } from "node:fs/promises"
+import { buildFlowAgentInstaller } from "./traffic-flow-agent-install.js"
 
 export async function listTrafficFlowHostFiles() {
   const row = await getTrafficFlowSettingsRow()
-  return buildTrafficFlowHostFiles({ ...row, peers: parseJsonArray(row.peersJson) })
+  const files = buildTrafficFlowHostFiles({ ...row, peers: parseJsonArray(row.peersJson) })
+  // The same source is used by a repository install and by the downloadable self-contained installer.
+  const installer = await readFile(new URL("../../../deploy/netflow-agent/install.sh", import.meta.url), "utf8")
+  const conf = files.find(file => file.id === "wg-quick")!.code
+  const code = buildFlowAgentInstaller(conf, row.flowListenPort, installer)
+  return [{ id: "agent", label: "Агент Docker", filename: "install-netflow-agent.sh", code }, ...files]
 }

backend/src/services/traffic-flow-operation-queue.test.ts

@@ -0,0 +1,21 @@
+import assert from "node:assert/strict"
+import { FlowConfigurationQueue } from "./traffic-flow-operation-queue.js"
+const queue = new FlowConfigurationQueue()
+let release!: () => void, active = 0, maximum = 0
+const hold = new Promise<void>(resolve => { release = resolve })
+const events: number[] = []
+const tasks = Array.from({ length: 32 }, (_, i) => queue.run(async () => {
+  active++; maximum = Math.max(active, maximum); events.push(i)
+  try { if (i === 0) await hold; if (i === 1) throw Error("fixture failure"); return i }
+  finally { active-- }
+}))
+const done = Promise.allSettled(tasks)
+await new Promise(resolve => setImmediate(resolve))
+assert.deepEqual(events, [0]); assert.equal(active, 1)
+release()
+const results = await done
+assert.equal(maximum, 1); assert.equal(results[1].status, "rejected")
+assert.equal(results.filter(result => result.status === "fulfilled").length, 31)
+assert.deepEqual(events, Array.from({ length: 32 }, (_, i) => i))
+assert.equal(await queue.run(async () => "recovered"), "recovered")
+console.log("NetFlow queue: 32 concurrent operations use one lock waiter, failures release the queue")

backend/src/services/traffic-flow-operation-queue.ts

@@ -0,0 +1,12 @@
+/** Only one local operation may wait for the PostgreSQL advisory lock.
+ * Otherwise queued requests could exhaust the pool needed by its current owner.
+ */
+export class FlowConfigurationQueue {
+  private tail: Promise<void> = Promise.resolve()
+
+  run<T>(operation: () => Promise<T>): Promise<T> {
+    const result = this.tail.then(operation)
+    this.tail = result.then(() => undefined, () => undefined)
+    return result
+  }
+}

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

@@ -22,6 +22,8 @@ import {
 import { refreshFlowExporterMap, startTrafficFlowListener } from "./traffic-flow-ingest.js"
 import { listTrafficFlowHostFiles } from "./traffic-flow-host-files.js"
 import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js"
+import { agentSocket } from "./traffic-flow-agent-client.js"
+import { getFlowHostStatus, preflightFlowHost, reconcileFlowHost, withFlowConfigurationLock } from "./traffic-flow-host-agent.js"
 
 const IFACE_NAME = "wg-flow"
 const JH_LISTEN_PORT = 13232
@@ -202,8 +204,16 @@ export function usablePublicHost(raw: string | undefined): string {
 export async function applyFlowOverlay(
   serverIdRaw: string | number,
   opts?: { publicEndpoint?: string; requestHost?: string; disableGreFastPath?: boolean },
+): Promise<TrafficFlowOverlayResult> {
+  return withFlowConfigurationLock(() => applyFlowOverlayLocked(serverIdRaw, opts))
+}
+
+async function applyFlowOverlayLocked(
+  serverIdRaw: string | number,
+  opts?: { publicEndpoint?: string; requestHost?: string; disableGreFastPath?: boolean },
 ): Promise<TrafficFlowOverlayResult> {
   const steps: string[] = []
+  await preflightFlowHost()
   const keys = await ensureHostKeys()
   const settings = await getTrafficFlowSettingsRow()
   const hostPublicKey = settings.hostPublicKey || keys.publicKey
@@ -338,8 +348,21 @@ export async function applyFlowOverlay(
     await refreshFlowExporterMap()
     steps.push("Коллектор IPFIX на MM включён")
 
+    let hostStatus = undefined
+    let hostError: string | undefined
+    if (agentSocket()) {
+      try {
+        hostStatus = await reconcileFlowHost()
+        steps.push("Пир применён на Linux-хосте без перезапуска WireGuard")
+      } catch (error) {
+        hostError = error instanceof Error ? error.message : "Настройки хоста не применены"
+        hostStatus = await getFlowHostStatus()
+        steps.push("MikroTik настроен; требуется повторное применение настроек хоста")
+      }
+    }
+
     return {
-      ok: true,
+      ok: !hostError,
       serverId: server.id,
       interfaceName: IFACE_NAME,
       address,
@@ -347,11 +370,14 @@ export async function applyFlowOverlay(
       linuxPeerBlock: linuxPeerBlock(publicKey, address, server.name || server.host, endpointHost),
       trafficFlow: true,
       steps,
-      hostFiles: await listTrafficFlowHostFiles(),
+      hostFiles: agentSocket() ? [] : await listTrafficFlowHostFiles(),
+      stage: hostError ? "host-pending" : agentSocket() ? "applied" : "manual",
+      ...(hostStatus ? { hostStatus } : {}),
+      ...(hostError ? { hostError } : {}),
     }
   } catch (e) {
     const msg = e instanceof MikrotikError ? e.message : e instanceof Error ? e.message : String(e)
-    const err = Object.assign(new Error(`RouterOS: ${msg}`), { statusCode: 502 })
+    const err = Object.assign(new Error(`Настройка MikroTik могла быть выполнена частично: ${msg}. Проверьте состояние и повторите подключение.`), { statusCode: 502 })
     throw err
   }
 }