MikrotikManager/backend/src/services/traffic-flow-analytics.ts

import { eq } from "drizzle-orm"
import { db, dbAll } from "../db/index.js"
import { appUsers, userInterfaceBindings } from "../db/schema.js"
import type {
  FlowAnalyticsDto,
  FlowBreakdownRow,
  FlowClientsDto,
  FlowEntityCard,
  FlowExportersDto,
  FlowMapEdge,
  FlowMonthlyDto,
  FlowPathRow,
  FlowTalkerDto,
} from "@mmapp/contracts/traffic-flow"
import { protoName } from "./traffic-flow-parse.js"
import {
  getFlowListenerState,
  getFlowRuntimeCounters,
  getFlowWorkerHealth,
  getRingMbps,
  listFlowRowsForWindow,
  type PendingFlowRow,
} from "./traffic-flow-ingest.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"
import { dedupFlowRowsMaxBytes, flowTupleKey } from "./traffic-flow-dedup.js"
import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js"
import { classifyFlowDst, refreshFlowCatalogInBackground } from "./traffic-flow-classify.js"
import { isIsoCountry } from "./traffic-flow-brands.js"
import { classifyFlowPlane, flowBps, shouldKeepPlane } from "./traffic-flow-planes.js"
import { pickInternetPeer } from "./traffic-flow-ip.js"
import {
  enGreIfaceNames,
  getServerCatalog,
  latestWireBps,
  loadFlowTopology,
  resolveClient,
  resolveEn,
} from "./traffic-flow-topology.js"

export const LIVE_ANALYTICS_MINUTES = 5
const LIVE_DEGRADED_PENDING = Math.floor(MAX_PENDING * 0.8)

export interface FlowAnalyticsQuery {
  minutes: number
  serverId?: number
  userId?: string
  iface?: string
  /** Default true: один 5-tuple = max байт по ifaces. */
  dedup?: boolean
  /** Default true: скрыть GRE/WG между клиентами JH. */
  excludeMesh?: boolean
  /** Default true: скрыть overlay GRE/ESP JH↔EN из payload KPI. */
  excludeOverlay?: boolean
  skipHeavy?: boolean
}

function bpsToMbps(bps: number): number {
  return bps / 1_000_000
}

function topN(
  map: Map<string, { bytes: number; packets: number; label?: string }>,
  windowSec: number,
  n: number,
): FlowBreakdownRow[] {
  const total = [...map.values()].reduce((a, v) => a + v.bytes, 0) || 1
  return [...map.entries()]
    .sort((a, b) => b[1].bytes - a[1].bytes)
    .slice(0, n)
    .map(([id, v]) => ({
      id,
      label: v.label || id,
      bytes: v.bytes,
      packets: v.packets,
      bps: (v.bytes * 8) / windowSec,
      percent: (v.bytes / total) * 100,
    }))
}

function bump(
  map: Map<string, { bytes: number; packets: number; label?: string }>,
  id: string,
  bytes: number,
  packets: number,
  label?: string,
) {
  const prev = map.get(id) ?? { bytes: 0, packets: 0, label }
  prev.bytes += bytes
  prev.packets += packets
  if (label) prev.label = label
  map.set(id, prev)
}

async function userIfaceAllow(userId: string): Promise<Map<number, Set<string>> | null> {
  if (!userId) return null
  const binds = await db.select().from(userInterfaceBindings).where(eq(userInterfaceBindings.userId, userId))
  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 seriesFromRows(rows: PendingFlowRow[], minutes: number): { rx: number[]; tx: number[] } {
  const slots = Math.min(60, Math.max(5, minutes))
  const slotMs = (minutes * 60_000) / slots
  const start = Date.now() - minutes * 60_000
  const rx = Array(slots).fill(0) as number[]
  const tx = Array(slots).fill(0) as number[]
  for (const r of rows) {
    const t = Date.parse(r.bucketAt)
    if (!Number.isFinite(t)) continue
    const idx = Math.min(slots - 1, Math.max(0, Math.floor((t - start) / slotMs)))
    rx[idx] += r.bytes
  }
  const slotSec = Math.max(1, slotMs / 1000)
  return {
    rx: rx.map((b) => bpsToMbps((b * 8) / slotSec)),
    tx,
  }
}

function snapshotStatus(serverId: number): FlowEntityCard["status"] {
  void serverId
  return "online"
}

function topLabel(map: Map<string, { bytes: number; packets: number; label?: string }>, fallback = "—"): string {
  let best = fallback
  let bestBytes = 0
  for (const [id, v] of map) {
    if (v.bytes > bestBytes) {
      bestBytes = v.bytes
      best = v.label || id
    }
  }
  return best
}

export async function buildFlowAnalytics(q: FlowAnalyticsQuery): Promise<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 = await 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

async function buildFlowAnalyticsUncached(q: FlowAnalyticsQuery): Promise<FlowAnalyticsDto> {
  const settings = await getTrafficFlowSettingsRow()
  const top = Math.min(50, Math.max(10, settings.topN))
  const windowSec = Math.max(60, q.minutes * 60)
  const raw = await listFlowRowsForWindow(q.minutes)
  const allow = q.userId ? await userIfaceAllow(q.userId) : null
  const catalog = await 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
  const excludeOverlay = q.excludeOverlay !== false
  const topo = await loadFlowTopology()

  refreshFlowCatalogInBackground()

  const applications = new Map<string, { bytes: number; packets: number; label?: string }>()
  const protocols = new Map<string, { bytes: number; packets: number; label?: string }>()
  const sources = new Map<string, { bytes: number; packets: number; label?: string }>()
  const destinations = new Map<string, { bytes: number; packets: number; label?: string }>()
  const ifacesMap = new Map<string, { bytes: number; packets: number; index: string }>()
  const asns = new Map<string, { bytes: number; packets: number; label?: string }>()
  const countries = new Map<string, { bytes: number; packets: number; label?: string }>()
  const categories = new Map<string, { bytes: number; packets: number; label?: string }>()
  const services = new Map<string, { bytes: number; packets: number; label?: string }>()
  const conv = new Map<string, FlowTalkerDto & { rawBytes: number; flowStartMs: number; flowEndMs: number }>()
  const edgeAcc = new Map<string, FlowMapEdge & { catBytes: Map<string, number> }>()
  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
  let bytesOverlay = 0
  let bytesMesh = 0
  const ifacesForWire = new Set<string>()

  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
    ifacesForWire.add(resolved.name)
    if (outResolved.name && outResolved.name !== "—") ifacesForWire.add(outResolved.name)
    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 (plane === "payload") bytesPayload += r.bytes
    else if (plane === "overlay") bytesOverlay += r.bytes
    else if (plane === "client_mesh") bytesMesh += r.bytes
    if (!shouldKeepPlane(plane, { excludeMesh, excludeOverlay })) continue
    matched.push(r)

    const ifaceKey = resolved.name
    const prevIf = ifacesMap.get(ifaceKey) ?? { bytes: 0, packets: 0, index: resolved.index }
    prevIf.bytes += r.bytes
    prevIf.packets += r.packets
    ifacesMap.set(ifaceKey, prevIf)
  }

  const working = wantDedup ? dedupFlowRowsMaxBytes(matched) : matched
  const conversationsRaw = new Set(matched.map((r) => `${flowTupleKey(r)}|${r.inIface}`)).size

  let totalBytes = 0
  let totalPackets = 0
  for (const r of working) {
    const resolved = resolveIfaceName(r.serverId, r.inIface)
    totalBytes += r.bytes
    totalPackets += r.packets
    srcs.add(r.src)
    dsts.add(r.dst)
    const 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(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)
    bump(destinations, r.dst, r.bytes, r.packets)
    bump(categories, classified.category, r.bytes, r.packets)
    bump(services, classified.service, r.bytes, r.packets)
    if (ripe?.ok && ripe.asn) {
      const asnId = String(ripe.asn)
      const asnLabel = ripe.holder ? `AS${ripe.asn} ${ripe.holder}` : `AS${ripe.asn}`
      bump(asns, asnId, r.bytes, r.packets, asnLabel)
    }
    const dstCountry = ripe?.ok && isIsoCountry(ripe.country) ? ripe.country : ""
    if (dstCountry) {
      bump(countries, dstCountry, r.bytes, r.packets)
    }

    if (!skipHeavy) {
      const outResolved = resolveIfaceName(r.serverId, r.outIface)
      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)
      const client = resolveClient(topo, r.serverId, resolved.name)
      const en = resolveEn(topo, r.nextHop, outResolved.name)
      const ckey = wantDedup
        ? flowTupleKey(r)
        : `${flowTupleKey(r)}|${r.inIface}`
      const prev = conv.get(ckey)
      if (prev) {
        prev.rawBytes += r.bytes
        prev.bytes += r.bytes
        prev.packets += r.packets
        if (r.flowStartMs && (!prev.flowStartMs || r.flowStartMs < prev.flowStartMs)) prev.flowStartMs = r.flowStartMs
        if (r.flowEndMs > prev.flowEndMs) prev.flowEndMs = r.flowEndMs
      } else {
        conv.set(ckey, {
          serverId: String(r.serverId),
          serverName: nameById.get(r.serverId) ?? String(r.serverId),
          src: r.src,
          dst: r.dst,
          proto: r.proto,
          protoName: protoName(r.proto),
          srcPort: r.srcPort,
          dstPort: r.dstPort,
          bytes: r.bytes,
          packets: r.packets,
          bps: 0,
          inIface: resolved.name,
          inIfaceIndex: resolved.index,
          outIface: outResolved.name !== "—" ? outResolved.name : undefined,
          nextHop: r.nextHop || undefined,
          application: app,
          category: classified.category,
          service: classified.service,
          dstCountry: dstCountry || undefined,
          dstAsn: ripe?.asn || undefined,
          clientId: client?.userId,
          clientName: client?.name,
          enId: en ? String(en.id) : undefined,
          enName: en?.name,
          plane,
          rawBytes: r.bytes,
          flowStartMs: r.flowStartMs ?? 0,
          flowEndMs: r.flowEndMs ?? 0,
        })
      }

      const pathKey = `${client?.userId || "unknown"}|${r.serverId}|${en?.id || ""}|${r.dst}|${resolved.name}`
      const pathPrev = pathAcc.get(pathKey)
      if (pathPrev) {
        pathPrev.bytes += r.bytes
        pathPrev.packets += r.packets
      } else {
        pathAcc.set(pathKey, {
          id: pathKey,
          clientId: client?.userId || "unknown",
          clientName: client?.name || "Неизвестный клиент",
          ifaces: client ? [...(topo.clientIfaces.get(r.serverId) ?? [resolved.name])].join(", ") : resolved.name,
          serverId: String(r.serverId),
          serverName: nameById.get(r.serverId) ?? String(r.serverId),
          inIface: resolved.name,
          outIface: outResolved.name !== "—" ? outResolved.name : "",
          enId: en ? String(en.id) : "",
          enName: en?.name || "",
          dst: r.dst,
          service: classified.service,
          category: classified.category,
          plane,
          bytes: r.bytes,
          packets: r.packets,
          bps: 0,
        })
      }

      const toCountry = dstCountry
      if (toCountry) {
        const fromCountry = countryById.get(r.serverId) || "UN"
        const ekey = `${r.serverId}|${toCountry}`
        let edge = edgeAcc.get(ekey)
        if (!edge) {
          edge = {
            fromId: String(r.serverId),
            fromLabel: nameById.get(r.serverId) ?? String(r.serverId),
            fromCountry,
            toCountry,
            toAsn: ripe?.asn ?? 0,
            category: classified.category,
            bytes: 0,
            bps: 0,
            catBytes: new Map(),
          }
          edgeAcc.set(ekey, edge)
        }
        edge.bytes += r.bytes
        if (ripe?.asn) edge.toAsn = ripe.asn
        edge.catBytes.set(classified.category, (edge.catBytes.get(classified.category) ?? 0) + r.bytes)
      }
    }
  }

  enqueueRipeMisses(peers)

  const conversationsList = [...conv.values()]
    .map((t) => {
      const { rawBytes, flowStartMs, flowEndMs, ...rest } = t
      return { ...rest, bps: flowBps(rawBytes, flowStartMs, flowEndMs, windowSec) }
    })
    .sort((a, b) => b.bytes - a.bytes)
    .slice(0, top)

  const paths: FlowPathRow[] = [...pathAcc.values()]
    .map((p) => ({ ...p, bps: (p.bytes * 8) / windowSec }))
    .sort((a, b) => b.bytes - a.bytes)
    .slice(0, top)

  const topProto = topLabel(protocols)
  const topCategory = topLabel(categories)

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

  const fromBuckets = seriesFromRows(matched, q.minutes)
  const rxSeries = q.minutes <= 15 ? ring.rx : fromBuckets.rx
  const txSeries = q.minutes <= 15 ? ring.tx : fromBuckets.tx

  const ifaceRows = [...ifacesMap.entries()]
    .sort((a, b) => b[1].bytes - a[1].bytes)
    .map(([name, v]) => ({
      name,
      index: v.index,
      bps: (v.bytes * 8) / windowSec,
    }))

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

  const overlayRing = ringServer
    ? getRingMbps(ringServer, RING_OVERLAY)
    : { rxNow: 0, txNow: 0 }
  const greNames = ringServer ? enGreIfaceNames(topo, ringServer, [...ifacesForWire]) : []
  const wire = ringServer ? await latestWireBps(ringServer, greNames) : { bps: 0, bytes: 0 }

  return {
    bpsNow: (ring.rxNow + ring.txNow) * 1_000_000 || (totalBytes * 8) / windowSec,
    bytes: totalBytes,
    packets: totalPackets,
    conversations: conv.size,
    conversationsRaw,
    uniqueSrc: srcs.size,
    uniqueDst: dsts.size,
    topProto,
    topCategory,
    rxSeries,
    txSeries,
    applications: topN(applications, windowSec, top),
    protocols: topN(protocols, windowSec, top),
    sources: topN(sources, windowSec, top),
    destinations: topN(destinations, windowSec, top),
    interfaces: [...ifacesMap.entries()].map(([label, v]) => ({
      id: label,
      label,
      bytes: v.bytes,
      packets: v.packets,
      bps: (v.bytes * 8) / windowSec,
      percent: (v.bytes / ifaceRawBytes) * 100,
    })).sort((a, b) => b.bytes - a.bytes),
    asns: topN(asns, windowSec, top),
    countries: topN(countries, windowSec, top),
    categories: topN(categories, windowSec, top),
    services: topN(services, windowSec, top),
    mapEdges,
    conversationsList,
    paths,
    ifaces: ifaceRows,
    live: listener.bound,
    dedupApplied: wantDedup,
    degraded: skipHeavy,
    bytesPayload,
    bytesOverlay,
    bytesMesh,
    bytesWire: wire.bytes,
    bpsOverlay: (overlayRing.rxNow + overlayRing.txNow) * 1_000_000 || (bytesOverlay * 8) / windowSec,
    bpsWire: wire.bps,
    excludeMeshApplied: excludeMesh,
    excludeOverlayApplied: excludeOverlay,
  }
}

function summarizeByServer(rows: PendingFlowRow[]) {
  const bytes = new Map<number, number>()
  const sessions = new Map<number, number>()
  for (const r of rows) {
    bytes.set(r.serverId, (bytes.get(r.serverId) ?? 0) + r.bytes)
    sessions.set(r.serverId, (sessions.get(r.serverId) ?? 0) + 1)
  }
  return { bytes, sessions }
}

export async function listFlowExporters(minutes: number): Promise<FlowExportersDto> {
  const runtime = await getFlowRuntimeCounters()
  const rows = await listFlowRowsForWindow(minutes)
  const { bytes, sessions } = summarizeByServer(rows)
  const ids = new Set<number>([...bytes.keys()])
  for (const p of await listHostPeers()) ids.add(p.serverId)
  const catalog = await getServerCatalog()
  const emptySeries = Array(60).fill(0) as number[]
  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,
        subtitle: s.host,
        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,
        sessions: sessions.get(s.id) ?? 0,
        rxSeries: ring.rx.some((v) => v > 0) ? ring.rx : emptySeries,
        txSeries: ring.tx,
        bytes: total,
      } satisfies FlowEntityCard
    })
    .sort((a, b) => b.rxNow - a.rxNow)
  const listener = getFlowListenerState()
  return {
    exporters,
    lastExporterIp: runtime.lastExporterIp,
    lastError: runtime.lastError,
    packetsReceived: runtime.packetsReceived,
    lastDatagramAt: runtime.lastDatagramAt,
    listenerBound: listener.bound,
    listenerAddress: listener.address,
  }
}

export async function listFlowClients(minutes: number): Promise<{ clients: { id: string; name: string; subtitle: string; site: string; country: string; status: "online" | "offline" | "degraded"; rxNow: number; txNow: number; sessions: number; rxSeries: number[]; txSeries: number[]; bytes: number; }[]; }> {
  const users = await db.select().from(appUsers)
  const binds = await db.select().from(userInterfaceBindings)
  const byUser = new Map<string, typeof binds>()
  for (const b of binds) {
    const list = byUser.get(b.userId) ?? []
    list.push(b)
    byUser.set(b.userId, list)
  }
  const rows = await listFlowRowsForWindow(minutes)
  const emptySeries = Array(60).fill(0) as number[]
  const windowSec = Math.max(60, minutes * 60)
  const clients: FlowEntityCard[] = []
  for (const u of users) {
    const userBinds = byUser.get(u.id) ?? []
    if (userBinds.length === 0) continue
    const allow = new Map<number, Set<string>>()
    for (const b of userBinds) {
      const set = allow.get(b.serverId) ?? new Set<string>()
      set.add(b.interfaceName)
      allow.set(b.serverId, set)
    }
    let total = 0
    let sessions = 0
    for (const r of rows) {
      const resolved = resolveIfaceName(r.serverId, r.inIface)
      const names = allow.get(r.serverId)
      if (!names) continue
      if (!names.has(resolved.name) && !names.has(r.inIface)) continue
      total += r.bytes
      sessions += 1
    }
    const firstServer = userBinds[0]?.serverId
    const ring = firstServer ? getRingMbps(firstServer, "__all__") : { rx: emptySeries, tx: emptySeries, rxNow: 0, txNow: 0 }
    clients.push({
      id: u.id,
      name: u.login,
      subtitle: u.name || u.login,
      site: `${userBinds.length} ifaces`,
      country: "UN",
      status: u.active ? "online" : "offline",
      rxNow: (total * 8) / windowSec / 1_000_000 || ring.rxNow,
      txNow: ring.txNow,
      sessions,
      rxSeries: ring.rx.some((v) => v > 0) ? ring.rx : emptySeries,
      txSeries: ring.tx,
      bytes: total,
    })
  }
  clients.sort((a, b) => b.rxNow - a.rxNow)
  return { clients }
}

export async function formatLiveSseFromBuilder(
  build: () => unknown | Promise<unknown>,
): Promise<{ event: "sample" | "error"; data: unknown }> {
  try {
    const data = await build()
    return { event: "sample", data }
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err)
    return { event: "error", data: { error: message } }
  }
}

export function isFlowAnalyticsDegraded(): boolean {
  const health = getFlowWorkerHealth()
  return health.pendingSize >= LIVE_DEGRADED_PENDING
}

export async function safeBuildLiveFlowSample(q: Omit<FlowAnalyticsQuery, "minutes" | "skipHeavy">): Promise<{
  event: "sample" | "error"
  data: unknown
}> {
  return formatLiveSseFromBuilder(async () => {
    const skipHeavy = isFlowAnalyticsDegraded()
    return await buildFlowAnalytics({ ...q, minutes: LIVE_ANALYTICS_MINUTES, skipHeavy })
  })
}

function monthBounds(month: string): { start: string; end: string } | null {
  if (!/^\d{4}-\d{2}$/.test(month)) return null
  const [yearRaw, monthRaw] = month.split("-")
  const year = Number(yearRaw)
  const monthIdx = Number(monthRaw)
  if (!Number.isFinite(year) || monthIdx < 1 || monthIdx > 12) return null
  const start = `${month}-01`
  const endDate = new Date(Date.UTC(year, monthIdx, 1))
  const end = endDate.toISOString().slice(0, 10)
  return { start, end }
}

function toBreakdown(
  rows: Array<{ key: string; bytes: number; packets: number }>,
  totalBytes: number,
  windowSec: number,
): FlowBreakdownRow[] {
  const denom = totalBytes || 1
  return rows
    .sort((a, b) => b.bytes - a.bytes)
    .map((r) => ({
      id: r.key,
      label: r.key,
      bytes: r.bytes,
      packets: r.packets,
      bps: (r.bytes * 8) / windowSec,
      percent: (r.bytes / denom) * 100,
    }))
}

export async function getFlowMonthly(month: string, serverId?: number): Promise<FlowMonthlyDto> {
  const bounds = monthBounds(month)
  if (!bounds) {
    return { month, bytes: 0, countries: [], services: [], asns: [] }
  }
  const params: Array<string | number> = [bounds.start, bounds.end]
  let where = "day >= ? AND day < ? AND dim IN ('country', 'service', 'asn')"
  if (serverId != null) {
    where += " AND server_id = ?"
    params.push(serverId)
  }
  const rows = await dbAll<{ dim: string; key: string; bytes: number; packets: number }>(`
    SELECT dim AS dim, key AS key, SUM(bytes) AS bytes, SUM(packets) AS packets
    FROM flow_daily_dims
    WHERE ${where}
    GROUP BY dim, key
  `, params)

  const countries: Array<{ key: string; bytes: number; packets: number }> = []
  const services: Array<{ key: string; bytes: number; packets: number }> = []
  const asns: Array<{ key: string; bytes: number; packets: number }> = []
  let bytes = 0
  for (const row of rows) {
    const rec = { key: row.key, bytes: Number(row.bytes) || 0, packets: Number(row.packets) || 0 }
    if (row.dim === "country") {
      countries.push(rec)
      bytes += rec.bytes
    } else if (row.dim === "service") services.push(rec)
    else if (row.dim === "asn") asns.push(rec)
  }
  const daysInMonth = Math.max(1, Math.round((Date.parse(`${bounds.end}T00:00:00Z`) - Date.parse(`${bounds.start}T00:00:00Z`)) / 86_400_000))
  const windowSec = daysInMonth * 86_400
  const countryTotal = countries.reduce((a, r) => a + r.bytes, 0) || bytes || 1
  return {
    month,
    bytes,
    countries: toBreakdown(countries, countryTotal, windowSec),
    services: toBreakdown(services, services.reduce((a, r) => a + r.bytes, 0) || 1, windowSec),
    asns: toBreakdown(asns, asns.reduce((a, r) => a + r.bytes, 0) || 1, windowSec),
  }
}