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

import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
import type { FastifyReply, FastifyRequest } from "fastify"
import { env } from "../config.js"
import {
  trafficFlowOverlayRequestSchema,
  trafficFlowSettingsPatchSchema,
} from "@mmapp/contracts/traffic-flow"
import {
  ensureHostKeys,
  getTrafficFlowSettingsRow,
  toTrafficFlowSettingsDto,
  updateTrafficFlowSettings,
} from "../services/traffic-flow-settings.js"
import {
  getFlowListenerState,
  purgeTrafficFlowStore,
  startTrafficFlowListener,
  listFlowTalkers,
} from "../services/traffic-flow-ingest.js"
import {
  buildFlowAnalytics,
  getFlowMonthly,
  listFlowClients,
  listFlowExporters,
  safeBuildLiveFlowSample,
} from "../services/traffic-flow-analytics.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"

const LIVE_TICK_MS = 2000
export const MAX_FLOW_LIVE_SUBSCRIBERS = 4
let liveSubscribers = 0

export function tryAcquireFlowLiveSlot(): boolean {
  if (liveSubscribers >= MAX_FLOW_LIVE_SUBSCRIBERS) return false
  liveSubscribers += 1
  return true
}

export function releaseFlowLiveSlot(): void {
  liveSubscribers = Math.max(0, liveSubscribers - 1)
}

export function resetFlowLiveSlotsForTests(): void {
  liveSubscribers = 0
}

function rangeToMinutes(range: string | undefined): number {
  switch ((range ?? "5m").toLowerCase()) {
    case "5m": return 5
    case "15m": return 15
    case "1h": return 60
    case "4h": return 240
    case "24h": return 1440
    case "30d": return 1440
    default: return 5
  }
}

function parseId(raw: unknown): number | undefined {
  if (raw == null || raw === "") return undefined
  const n = Number.parseInt(String(raw), 10)
  return Number.isFinite(n) ? n : undefined
}

function parseDedup(raw: unknown): boolean {
  if (raw == null || raw === "") return true
  const s = String(raw).toLowerCase()
  return s !== "0" && s !== "false" && s !== "off"
}

function analyticsQuery(req: FastifyRequest) {
  const q = req.query as {
    range?: string
    serverId?: string
    userId?: string
    iface?: string
    dedup?: string
    excludeMesh?: string
    excludeOverlay?: string
  }
  return {
    minutes: rangeToMinutes(q.range),
    serverId: parseId(q.serverId),
    userId: q.userId?.trim() || undefined,
    iface: q.iface?.trim() || undefined,
    dedup: parseDedup(q.dedup),
    excludeMesh: parseDedup(q.excludeMesh),
    excludeOverlay: parseDedup(q.excludeOverlay),
  }
}

async function sendFlowTalkers(req: FastifyRequest, reply: FastifyReply) {
  const q = req.query as { range?: string }
  return reply.send(listFlowTalkers(rangeToMinutes(q.range)))
}

function requestPublicHost(req: FastifyRequest): string {
  const forwarded = req.headers["x-forwarded-host"]
  const raw = Array.isArray(forwarded) ? forwarded[0] : forwarded
  return raw || req.hostname || ""
}

async function applyOverlayHandler(req: FastifyRequest, reply: FastifyReply) {
  const parsed = trafficFlowOverlayRequestSchema.safeParse(req.body ?? {})
  if (!parsed.success) {
    return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
  }
  try {
    const result = await applyFlowOverlay(parsed.data.serverId, {
      publicEndpoint: parsed.data.publicEndpoint,
      requestHost: requestPublicHost(req),
    })
    return reply.send(result)
  } catch (e) {
    const status = (e as { statusCode?: number }).statusCode ?? 502
    const msg = e instanceof Error ? e.message : String(e)
    return reply.status(status).send({ error: msg })
  }
}

function writeSse(raw: NodeJS.WritableStream, event: string, data: unknown) {
  raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
}

function sleep(ms: number, signal: AbortSignal): Promise<void> {
  return new Promise((resolve, reject) => {
    if (signal.aborted) {
      reject(new Error("aborted"))
      return
    }
    const timer = setTimeout(() => {
      signal.removeEventListener("abort", onAbort)
      resolve()
    }, ms)
    const onAbort = () => {
      clearTimeout(timer)
      reject(new Error("aborted"))
    }
    signal.addEventListener("abort", onAbort, { once: true })
  })
}

const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => {
  app.get("/traffic/flow/settings", async (_req, reply) => {
    return reply.send(toTrafficFlowSettingsDto(getFlowListenerState()))
  })

  app.put("/traffic/flow/settings", async (req, reply) => {
    const parsed = trafficFlowSettingsPatchSchema.safeParse(req.body ?? {})
    if (!parsed.success) {
      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
    }
    updateTrafficFlowSettings(parsed.data)
    startTrafficFlowListener()
    return reply.send({ ok: true, settings: toTrafficFlowSettingsDto(getFlowListenerState()) })
  })

  app.post("/traffic/flow/settings/generate-keys", async (_req, reply) => {
    const result = ensureHostKeys()
    return reply.send({
      ok: true,
      created: result.created,
      publicKey: result.publicKey,
      settings: toTrafficFlowSettingsDto(getFlowListenerState()),
    })
  })

  app.get("/traffic/flow/host-files", async (_req, reply) => {
    const row = getTrafficFlowSettingsRow()
    if (!row.hostPrivateKey) ensureHostKeys()
    return reply.send({ files: listTrafficFlowHostFiles() })
  })

  app.post("/traffic/flow/purge", async (_req, reply) => {
    try {
      const result = await purgeTrafficFlowStore()
      appendEvent({
        level: "warning",
        eventType: "traffic.flow.purge",
        sourceModule: "traffic",
        title: "Сброшены данные NetFlow",
        message: `Удалены сессии ${result.deleted.buckets}, minute ${result.deleted.minuteStats}, daily ${result.deleted.dailyDims}`,
        entityType: "traffic_flow",
        entityId: "purge",
        payload: {
          buckets: result.deleted.buckets,
          minuteStats: result.deleted.minuteStats,
          minuteDims: result.deleted.minuteDims,
          dailyDims: result.deleted.dailyDims,
          fileBytesBefore: result.fileBytesBefore,
          fileBytesAfter: result.fileBytesAfter,
          vacuumed: result.vacuumed,
        },
      })
      return reply.send(result)
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err)
      const status = message.includes("уже выполняется") ? 409 : 500
      return reply.status(status).send({ error: message })
    }
  })

  app.post("/traffic/flow/overlay", applyOverlayHandler)
  app.post("/traffic/flow-overlay", applyOverlayHandler)

  app.get("/traffic/flow", sendFlowTalkers)
  app.get("/traffic/flows", sendFlowTalkers)

  app.get("/traffic/flow/exporters", async (req, reply) => {
    const q = req.query as { range?: string }
    return reply.send(listFlowExporters(rangeToMinutes(q.range)))
  })

  app.get("/traffic/flow/clients", async (req, reply) => {
    const q = req.query as { range?: string }
    return reply.send(listFlowClients(rangeToMinutes(q.range)))
  })

  app.get("/traffic/flow/analytics", async (req, reply) => {
    return reply.send(buildFlowAnalytics(analyticsQuery(req)))
  })

  app.get("/traffic/flow/monthly", async (req, reply) => {
    const q = req.query as { month?: string; serverId?: string }
    const now = new Date()
    const month = /^\d{4}-\d{2}$/.test(q.month ?? "")
      ? (q.month as string)
      : `${now.getUTCFullYear()}-${String(now.getUTCMonth() + 1).padStart(2, "0")}`
    return reply.send(getFlowMonthly(month, parseId(q.serverId)))
  })

  app.get("/traffic/flow/live", async (req, reply) => {
    if (!tryAcquireFlowLiveSlot()) {
      return reply.status(429).send({ error: "Слишком много live-подписок" })
    }
    const query = analyticsQuery(req)
    const liveQuery = {
      serverId: query.serverId,
      userId: query.userId,
      iface: query.iface,
      dedup: query.dedup,
      excludeMesh: query.excludeMesh,
      excludeOverlay: query.excludeOverlay,
    }
    const abort = new AbortController()
    const onClose = () => abort.abort()
    req.raw.on("close", onClose)

    reply.hijack()
    req.raw.setTimeout(0)
    reply.raw.setTimeout(0)
    const origin = typeof req.headers.origin === "string" ? req.headers.origin : ""
    const allowed = env.CORS_ORIGIN
    const sseHeaders: Record<string, string> = {
      "Content-Type": "text/event-stream; charset=utf-8",
      "Cache-Control": "no-cache, no-transform",
      Connection: "keep-alive",
      "X-Accel-Buffering": "no",
    }
    if (origin && (allowed === "*" || allowed === origin)) {
      sseHeaders["Access-Control-Allow-Origin"] = origin
      sseHeaders["Access-Control-Allow-Credentials"] = "true"
      sseHeaders["Access-Control-Allow-Headers"] = "Authorization, Accept"
      sseHeaders.Vary = "Origin"
    }
    reply.raw.writeHead(200, sseHeaders)
    reply.raw.write(":\n\n")

    try {
      while (!abort.signal.aborted) {
        const payload = safeBuildLiveFlowSample(liveQuery)
        writeSse(reply.raw, payload.event, payload.data)
        await sleep(LIVE_TICK_MS, abort.signal)
      }
    } catch {
      /* abort / disconnect */
    } finally {
      releaseFlowLiveSlot()
      req.raw.off("close", onClose)
      try {
        reply.raw.end()
      } catch {
        /* already closed */
      }
    }
  })
}

export default trafficFlowRoutes