import { dbAll, dbQuery, pool } from "../db/index.js"
import { ensurePartitionFor, specForParent } from "../db/partitions.js"
import { normalizeParsedFlow, parseFlowPacket, protoName, type ParsedFlow, type ParsedFlowInput } from "./traffic-flow-parse.js"
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 { enqueueRipeMisses, pruneRipeSqlite } from "./traffic-flow-ripe.js"
import { invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js"
import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js"
import { canonicalFactIface } from "./traffic-flow-ifindex.js"
import { shouldWriteFlowFact } from "./traffic-flow-facts-filter.js"
import { canonicalIp } from "./traffic-flow-ip.js"
import { resolveInternetDest } from "./traffic-flow-dest.js"
import {
  getServerCatalog,
  loadFlowTopology,
  peekFlowTopology,
  peekServerCatalog,
} from "./traffic-flow-topology.js"
import {
  bumpFlowFact,
  factsPendingSize,
  flushFlowFacts,
  hourBucketIso,
  resetFactsForTests,
} from "./traffic-flow-facts.js"

export const TICK_MS = 2_000
export const PERSIST_MS = 10_000
export const STATS_PERSIST_MS = 15_000
export const RING_LEN = 60
export const MAX_PENDING = 50_000
export const DAILY_ASN_TOP = 500
export const DAILY_RETENTION_DAYS = 396
export const MINUTE_RETENTION_HOURS = 48

let pendingCap = MAX_PENDING
const ensuredParts = new Set<string>()

async function ensureParentPartition(parent: string, ts: string): Promise<void> {
  const spec = specForParent(parent)
  if (!spec) return
  const iso = ts.length === 10 ? `${ts}T00:00:00Z` : ts
  const key = `${parent}:${iso.slice(0, 10)}`
  if (ensuredParts.has(key)) return
  await ensurePartitionFor(pool, parent, spec.kind, new Date(iso))
  ensuredParts.add(key)
}

export interface PendingFlowRow {
  serverId: number
  bucketAt: string
  src: string
  dst: string
  proto: number
  srcPort: number
  dstPort: number
  bytes: number
  packets: number
  inIface: string
  outIface: string
  nextHop: string
  flowStartMs: number
  flowEndMs: number
  natSrc: string
  natDst: string
  natSrcPort: number
  natDstPort: number
}

function inetOrNull(value: string | null | undefined): string | null {
  const s = canonicalIp(value)
  return s.length > 0 ? s : null
}

export function isValidFlowInet(value: string): boolean {
  const s = canonicalIp(value)
  if (!s) return false
  const v4 = /^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/.exec(s)
  if (v4) {
    return v4.slice(1).every((octet) => {
      const n = Number(octet)
      return Number.isInteger(n) && n >= 0 && n <= 255
    })
  }
  if (!s.includes(":")) return false
  if (!/^[0-9a-fA-F:]+$/.test(s)) return false
  const parts = s.split(":")
  if (parts.length < 3 || parts.length > 8) return false
  return parts.every((p) => p.length <= 4)
}

function clampProto(n: number): number {
  if (!Number.isFinite(n)) return 0
  return Math.max(0, Math.min(255, Math.trunc(n)))
}

function clampPort(n: number): number {
  if (!Number.isFinite(n)) return 0
  return Math.max(0, Math.min(65535, Math.trunc(n)))
}

function sanitizeNatIp(value: string | null | undefined): string {
  const s = canonicalIp(value)
  if (!s || s === "0.0.0.0") return ""
  return isValidFlowInet(s) ? s : ""
}

function sanitizeFlowRow(r: PendingFlowRow): PendingFlowRow | null {
  const src = canonicalIp(r.src) || "0.0.0.0"
  const dst = canonicalIp(r.dst) || "0.0.0.0"
  if (!isValidFlowInet(src) || !isValidFlowInet(dst)) return null
  const next = inetOrNull(canonicalIp(r.nextHop))
  return {
    ...r,
    src,
    dst,
    nextHop: next && isValidFlowInet(next) ? next : "",
    proto: clampProto(r.proto),
    natSrc: sanitizeNatIp(r.natSrc),
    natDst: sanitizeNatIp(r.natDst),
    natSrcPort: clampPort(r.natSrcPort),
    natDstPort: clampPort(r.natDstPort),
  }
}

function flowUpsertParams(r: PendingFlowRow) {
  return {
    serverId: r.serverId,
    bucketAt: r.bucketAt,
    src: r.src,
    dst: r.dst,
    proto: r.proto,
    srcPort: r.srcPort,
    dstPort: r.dstPort,
    bytes: r.bytes,
    packets: r.packets,
    inIface: r.inIface,
    outIface: r.outIface,
    nextHop: inetOrNull(r.nextHop),
    flowStartMs: r.flowStartMs,
    flowEndMs: r.flowEndMs,
    natSrc: inetOrNull(r.natSrc),
    natDst: inetOrNull(r.natDst),
    natSrcPort: r.natSrcPort,
    natDstPort: r.natDstPort,
  }
}

export interface EngineStats {
  packetsReceived: number
  lastExporterIp: string | null
  lastError: string
  lastDatagramAt: string | null
  dropped: number
  rowsStored: number
  pendingSize: number
}

interface PendingEntry {
  serverId: number
  bucketAt: string
  flow: ParsedFlow
  bytes: number
  packets: number
}

interface MinuteRollup {
  bytes: number
  packets: number
  srcs: Set<string>
  dsts: Set<string>
  conversations: number
}

interface DimAcc {
  bytes: number
  packets: number
}

export interface ExporterResolveCtx {
  overlayPrefix: string
  byTunnelIp: Map<string, number>
  peers: OverlayPeerRef[]
  hostIps: Map<string, number>
}

let topN = 200
let retentionHours = 24

const pending = new Map<string, PendingEntry>()
const recent = new Map<string, PendingFlowRow>()
const tickAccum = new Map<string, { inBytes: number; outBytes: number }>()
const rings = new Map<string, { inBps: number[]; outBps: number[] }>()
const minuteRollup = new Map<string, MinuteRollup>()
const minuteDims = new Map<string, DimAcc>()

let packetsReceived = 0
let lastExporterIp: string | null = null
let lastError = ""
let lastDatagramAt: string | null = null
let dropped = 0
let rowsStored = 0
let lastFlushUsedTransaction = false
let lastPruneAt = 0
let lastPassiveCheckpointAt = 0
let lastPersistAt = 0
let lastFlushedMinute = ""
let lastStatsPersistAt = 0
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

function bumpDataEpoch(): void {
  dataEpoch += 1
}

export function flowDataEpoch(): number {
  return dataEpoch
}

function nowIso(): string {
  return new Date().toISOString()
}

export function minuteBucketIso(at = Date.now()): string {
  const d = new Date(at)
  d.setSeconds(0, 0)
  return d.toISOString()
}

function dayKey(bucketAt: string): string {
  return bucketAt.slice(0, 10)
}

export const RING_PAYLOAD = "__all__"
export const RING_OVERLAY = "__overlay__"
export const RING_MESH = "__mesh__"

function ringKey(serverId: number, iface: string): string {
  return `${serverId}\0${iface || RING_PAYLOAD}`
}

function pendingKey(serverId: number, bucketAt: string, flow: ParsedFlow): string {
  return `${serverId}\0${bucketAt}\0${flow.src}\0${flow.dst}\0${flow.proto}\0${flow.srcPort}\0${flow.dstPort}\0${flow.inIface}`
}

function rowKey(row: PendingFlowRow): string {
  return `${row.serverId}|${row.bucketAt}|${row.src}|${row.dst}|${row.proto}|${row.srcPort}|${row.dstPort}|${row.inIface}`
}

function rollupKey(serverId: number, bucketAt: string): string {
  return `${serverId}\0${bucketAt}`
}

function dimKey(serverId: number, bucketAt: string, dim: string, key: string): string {
  return `${serverId}\0${bucketAt}\0${dim}\0${key}`
}

function bumpTick(key: string, inBytes: number, outBytes: number): void {
  const prev = tickAccum.get(key) ?? { inBytes: 0, outBytes: 0 }
  prev.inBytes += inBytes
  prev.outBytes += outBytes
  tickAccum.set(key, prev)
}

function addToTick(serverId: number, flow: ParsedFlow, bytes: number): void {
  const plane = classifyFlowPlaneLite(flow)
  if (plane === "mgmt") return
  const bucket = plane === "overlay" ? RING_OVERLAY : plane === "client_mesh" ? RING_MESH : RING_PAYLOAD
  bumpTick(ringKey(serverId, bucket), bytes, 0)
  if (flow.inIface) bumpTick(ringKey(serverId, flow.inIface), bytes, 0)
  if (flow.outIface && flow.outIface !== flow.inIface) bumpTick(ringKey(serverId, flow.outIface), 0, bytes)
}

function emptyRing(): { inBps: number[]; outBps: number[] } {
  return { inBps: Array(RING_LEN).fill(0), outBps: Array(RING_LEN).fill(0) }
}

function bumpDim(serverId: number, bucketAt: string, dim: string, key: string, bytes: number, packets: number): void {
  if (!key) return
  const k = dimKey(serverId, bucketAt, dim, key)
  const prev = minuteDims.get(k)
  if (prev) {
    prev.bytes += bytes
    prev.packets += packets
    return
  }
  minuteDims.set(k, { bytes, packets })
}

function bumpRollup(serverId: number, bucketAt: string, flow: ParsedFlow, bytes: number, packets: number): void {
  const k = rollupKey(serverId, bucketAt)
  let acc = minuteRollup.get(k)
  if (!acc) {
    acc = { bytes: 0, packets: 0, srcs: new Set(), dsts: new Set(), conversations: 0 }
    minuteRollup.set(k, acc)
  }
  acc.bytes += bytes
  acc.packets += packets
  if (flow.src) acc.srcs.add(flow.src)
  if (flow.dst) acc.dsts.add(flow.dst)
  acc.conversations += 1
}

export function attachEngineSqlite(_handle?: unknown): void {
  /* engine uses pg pool via dbQuery/dbAll */
}

export function setPendingCapForTests(n: number | null): void {
  pendingCap = n == null ? MAX_PENDING : Math.max(1, n)
}

export function configureEngine(opts: { topN?: number; retentionHours?: number }): void {
  if (opts.topN != null) topN = Math.max(20, opts.topN)
  if (opts.retentionHours != null) retentionHours = Math.max(1, opts.retentionHours)
}

export function setExporterResolveCtx(ctx: ExporterResolveCtx | null): void {
  exporterCtx = ctx
}

export function resolveServerId(exporterIp: string): number | null {
  if (!exporterCtx) return null
  return pickServerIdForExporter({
    exporterIp,
    overlayPrefix: exporterCtx.overlayPrefix,
    byTunnelIp: exporterCtx.byTunnelIp,
    peers: exporterCtx.peers,
    hostIps: exporterCtx.hostIps,
  })
}

export function bumpPacketMeta(exporterIp: string): void {
  packetsReceived += 1
  lastExporterIp = exporterIp
  lastDatagramAt = nowIso()
}

export function setEngineError(message: string): void {
  lastError = message
}

export function getEngineStats(): EngineStats {
  return {
    packetsReceived,
    lastExporterIp,
    lastError,
    lastDatagramAt,
    dropped,
    rowsStored,
    pendingSize: pending.size,
  }
}

export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): void {
  if (flows.length) bumpDataEpoch()
  const bucketAt = minuteBucketIso()
  const hourAt = hourBucketIso()
  const ripeMisses: string[] = []
  const topo = peekFlowTopology()
  const catalog = peekServerCatalog()
  if (!topo) void loadFlowTopology().catch(() => {})
  if (!catalog) void getServerCatalog().catch(() => {})
  const serverType = catalog?.byId.get(serverId)?.type
  for (const raw of flows) {
    const flow = normalizeParsedFlow(raw)
    addToTick(serverId, flow, flow.bytes)
    bumpRollup(serverId, bucketAt, flow, flow.bytes, flow.packets)
    const destMeta = resolveInternetDest({
      src: flow.src,
      dst: flow.dst,
      proto: flow.proto,
      srcPort: flow.srcPort,
      dstPort: flow.dstPort,
      serverId,
      inIface: flow.inIface,
      topo,
      natSrc: flow.natSrc,
      natDst: flow.natDst,
      natSrcPort: flow.natSrcPort,
      natDstPort: flow.natDstPort,
    })
    const ripe = destMeta.ripe
    if (destMeta.dest && !ripe) ripeMisses.push(destMeta.dest)
    const classified = destMeta.classified
    const app = applicationName(flow.proto, flow.dstPort, flow.srcPort)
    const country = destMeta.country
    const asnKey = destMeta.asn ? String(destMeta.asn) : "unknown"
    bumpDim(serverId, bucketAt, "proto", protoName(flow.proto), flow.bytes, flow.packets)
    bumpDim(serverId, bucketAt, "app", app, flow.bytes, flow.packets)
    bumpDim(serverId, bucketAt, "iface", flow.inIface || "__unknown__", flow.bytes, flow.packets)
    bumpDim(serverId, bucketAt, "category", classified.category, flow.bytes, flow.packets)
    bumpDim(serverId, bucketAt, "service", classified.service, flow.bytes, flow.packets)
    if (country) bumpDim(serverId, bucketAt, "country", country, flow.bytes, flow.packets)
    bumpDim(serverId, bucketAt, "asn", asnKey, flow.bytes, flow.packets)
    if (shouldWriteFlowFact({
      serverId,
      serverType,
      inIface: flow.inIface,
      outIface: flow.outIface,
      proto: flow.proto,
      srcPort: flow.srcPort,
      dstPort: flow.dstPort,
      src: flow.src,
      dst: flow.dst,
      topo,
      dest: destMeta.dest,
      natSrc: flow.natSrc,
      natDst: flow.natDst,
      natSrcPort: flow.natSrcPort,
      natDstPort: flow.natDstPort,
    })) {
      bumpFlowFact({
        serverId,
        bucketAt: hourAt,
        iface: canonicalFactIface(serverId, flow.inIface),
        country: country || "XX",
        service: classified.service,
        asn: destMeta.asn,
        bytes: flow.bytes,
        packets: flow.packets,
      })
    }

    const key = pendingKey(serverId, bucketAt, flow)
    const prev = pending.get(key)
    if (prev) {
      prev.bytes += flow.bytes
      prev.packets += flow.packets
      if (flow.outIface && !prev.flow.outIface) prev.flow.outIface = flow.outIface
      if (flow.nextHop && !prev.flow.nextHop) prev.flow.nextHop = flow.nextHop
      if (flow.natSrc && !prev.flow.natSrc) prev.flow.natSrc = flow.natSrc
      if (flow.natDst && !prev.flow.natDst) prev.flow.natDst = flow.natDst
      if (flow.natSrcPort && !prev.flow.natSrcPort) prev.flow.natSrcPort = flow.natSrcPort
      if (flow.natDstPort && !prev.flow.natDstPort) prev.flow.natDstPort = flow.natDstPort
      if (flow.flowStartMs && (!prev.flow.flowStartMs || flow.flowStartMs < prev.flow.flowStartMs)) {
        prev.flow.flowStartMs = flow.flowStartMs
      }
      if (flow.flowEndMs > (prev.flow.flowEndMs ?? 0)) prev.flow.flowEndMs = flow.flowEndMs
      continue
    }
    if (pending.size >= pendingCap) {
      dropped += 1
      continue
    }
    pending.set(key, {
      serverId,
      bucketAt,
      flow: { ...flow },
      bytes: flow.bytes,
      packets: flow.packets,
    })
  }
  if (ripeMisses.length) enqueueRipeMisses(ripeMisses)
}

export function ingestDatagram(msg: Buffer, exporterIp: string): boolean {
  bumpPacketMeta(exporterIp)
  const flows = parseFlowPacket(msg, exporterIp)
  if (!flows.length) return true
  const serverId = resolveServerId(exporterIp)
  if (serverId == null) {
    setEngineError(
      `IPFIX от ${exporterIp}: нет jump-host с адресом wg-flow. Docker SNAT (172.x) при нескольких JH не различим.`,
    )
    return false
  }
  setEngineError("")
  maybeRefreshIfaces(serverId)
  queueParsedFlows(serverId, flows)
  return true
}

function toPendingRow(row: PendingEntry): PendingFlowRow {
  const flow = normalizeParsedFlow(row.flow)
  return {
    serverId: row.serverId,
    bucketAt: row.bucketAt,
    src: flow.src || "0.0.0.0",
    dst: flow.dst || "0.0.0.0",
    proto: flow.proto,
    srcPort: flow.srcPort,
    dstPort: flow.dstPort,
    bytes: row.bytes,
    packets: row.packets,
    inIface: flow.inIface,
    outIface: flow.outIface,
    nextHop: flow.nextHop,
    flowStartMs: flow.flowStartMs,
    flowEndMs: flow.flowEndMs,
    natSrc: flow.natSrc,
    natDst: flow.natDst,
    natSrcPort: flow.natSrcPort,
    natDstPort: flow.natDstPort,
  }
}

function mergeInto(map: Map<string, PendingFlowRow>, row: PendingFlowRow): void {
  const key = rowKey(row)
  const prev = map.get(key)
  if (prev) {
    prev.bytes += row.bytes
    prev.packets += row.packets
    if (row.outIface && !prev.outIface) prev.outIface = row.outIface
    if (row.nextHop && !prev.nextHop) prev.nextHop = row.nextHop
    if (row.natSrc && !prev.natSrc) prev.natSrc = row.natSrc
    if (row.natDst && !prev.natDst) prev.natDst = row.natDst
    if (row.natSrcPort && !prev.natSrcPort) prev.natSrcPort = row.natSrcPort
    if (row.natDstPort && !prev.natDstPort) prev.natDstPort = row.natDstPort
    if (row.flowStartMs && (!prev.flowStartMs || row.flowStartMs < prev.flowStartMs)) prev.flowStartMs = row.flowStartMs
    if (row.flowEndMs > (prev.flowEndMs ?? 0)) prev.flowEndMs = row.flowEndMs
    return
  }
  map.set(key, { ...row })
}

function pruneRecent(sinceMs = Date.now() - LIVE_WINDOW_MS): void {
  const cutoff = new Date(sinceMs).toISOString()
  for (const [key, row] of recent) {
    if (row.bucketAt < cutoff) recent.delete(key)
  }
  while (recent.size > MAX_PENDING) {
    const first = recent.keys().next().value
    if (first == null) break
    recent.delete(first)
  }
}

export function peekPendingFlows(): PendingFlowRow[] {
  return [...pending.values()].map(toPendingRow)
}

export function listLiveFlowRows(sinceIso: string): PendingFlowRow[] {
  const merged = new Map<string, PendingFlowRow>()
  for (const row of recent.values()) {
    if (row.bucketAt < sinceIso) continue
    mergeInto(merged, row)
  }
  for (const row of peekPendingFlows()) {
    if (row.bucketAt < sinceIso) continue
    mergeInto(merged, row)
  }
  return [...merged.values()]
}

export function rollFlowRings(): void {
  const keys = new Set([...tickAccum.keys(), ...rings.keys()])
  const sec = TICK_MS / 1000
  for (const key of keys) {
    const acc = tickAccum.get(key) ?? { inBytes: 0, outBytes: 0 }
    tickAccum.delete(key)
    const inBps = (acc.inBytes * 8) / sec
    const outBps = (acc.outBytes * 8) / sec
    let ring = rings.get(key)
    if (!ring) {
      ring = emptyRing()
      rings.set(key, ring)
    }
    ring.inBps.push(inBps)
    ring.inBps.shift()
    ring.outBps.push(outBps)
    ring.outBps.shift()
    const silent = ring.inBps.every((v) => v === 0) && ring.outBps.every((v) => v === 0)
    if (silent && !tickAccum.has(key)) rings.delete(key)
  }
}

export function getRingMbps(serverId: number, iface = RING_PAYLOAD): {
  rx: number[]
  tx: number[]
  rxNow: number
  txNow: number
} {
  const ring = rings.get(ringKey(serverId, iface))
  const scale = 1_000_000
  if (!ring) {
    return { rx: Array(RING_LEN).fill(0), tx: Array(RING_LEN).fill(0), rxNow: 0, txNow: 0 }
  }
  return {
    rx: ring.inBps.map((b) => b / scale),
    tx: ring.outBps.map((b) => b / scale),
    rxNow: (ring.inBps[RING_LEN - 1] ?? 0) / scale,
    txNow: (ring.outBps[RING_LEN - 1] ?? 0) / scale,
  }
}

export function snapshotRings(): Array<{ key: string; inBps: number[]; outBps: number[] }> {
  return [...rings.entries()].map(([key, ring]) => ({
    key,
    inBps: [...ring.inBps],
    outBps: [...ring.outBps],
  }))
}

export function applyRingSnapshot(rows: Array<{ key: string; inBps: number[]; outBps: number[] }>): void {
  rings.clear()
  for (const row of rows) {
    rings.set(row.key, { inBps: row.inBps, outBps: row.outBps })
  }
}

async function persistListenerStats(force = false): Promise<boolean> {
  if (
    lastPersistedStats
    && lastPersistedStats.packetsReceived === packetsReceived
    && lastPersistedStats.lastDatagramAt === lastDatagramAt
    && lastPersistedStats.lastExporterIp === lastExporterIp
    && lastPersistedStats.lastError === lastError
  ) {
    return false
  }
  const errorChanged = lastPersistedStats?.lastError !== lastError
  const exporterChanged = lastPersistedStats?.lastExporterIp !== lastExporterIp
  const now = Date.now()
  if (!force && !errorChanged && !exporterChanged && now - lastStatsPersistAt < STATS_PERSIST_MS) {
    return false
  }
  await dbQuery(`
    UPDATE traffic_flow_settings
    SET packets_received = @packetsReceived,
        last_datagram_at = @lastDatagramAt,
        last_exporter_ip = @lastExporterIp,
        last_error = @lastError,
        updated_at = @updatedAt
    WHERE id = 1
  `, {
    packetsReceived,
    lastDatagramAt,
    lastExporterIp,
    lastError,
    updatedAt: nowIso(),
  })
  lastPersistedStats = {
    packetsReceived,
    lastDatagramAt,
    lastExporterIp,
    lastError,
  }
  lastStatsPersistAt = now
  invalidateTrafficFlowSettingsCache()
  return true
}

async function upsertMinuteAndDaily(): Promise<void> {
  for (const [k, acc] of minuteRollup) {
    const [serverIdRaw, bucketAt] = k.split("\0")
    await ensureParentPartition("flow_minute_stats", bucketAt ?? "")
    await dbQuery(`
      INSERT INTO flow_minute_stats (
        server_id, bucket_at, bytes, packets, unique_src, unique_dst, conversations
      ) VALUES (
        @serverId, @bucketAt, @bytes, @packets, @uniqueSrc, @uniqueDst, @conversations
      )
      ON CONFLICT(server_id, bucket_at) DO UPDATE SET
        bytes = flow_minute_stats.bytes + excluded.bytes,
        packets = flow_minute_stats.packets + excluded.packets,
        unique_src = GREATEST(flow_minute_stats.unique_src, excluded.unique_src),
        unique_dst = GREATEST(flow_minute_stats.unique_dst, excluded.unique_dst),
        conversations = flow_minute_stats.conversations + excluded.conversations
    `, {
      serverId: Number(serverIdRaw),
      bucketAt,
      bytes: acc.bytes,
      packets: acc.packets,
      uniqueSrc: acc.srcs.size,
      uniqueDst: acc.dsts.size,
      conversations: acc.conversations,
    })
  }
  for (const [k, acc] of minuteDims) {
    const [serverIdRaw, bucketAt, dim, key] = k.split("\0")
    await ensureParentPartition("flow_minute_dims", bucketAt ?? "")
    await ensureParentPartition("flow_daily_dims", dayKey(bucketAt ?? ""))
    await dbQuery(`
      INSERT INTO flow_minute_dims (server_id, bucket_at, dim, key, bytes, packets)
      VALUES (@serverId, @bucketAt, @dim, @key, @bytes, @packets)
      ON CONFLICT(server_id, bucket_at, dim, key) DO UPDATE SET
        bytes = flow_minute_dims.bytes + excluded.bytes,
        packets = flow_minute_dims.packets + excluded.packets
    `, {
      serverId: Number(serverIdRaw),
      bucketAt,
      dim,
      key,
      bytes: acc.bytes,
      packets: acc.packets,
    })
    if (dim === "country" || dim === "service" || dim === "asn") {
      await dbQuery(`
        INSERT INTO flow_daily_dims (server_id, day, dim, key, bytes, packets)
        VALUES (@serverId, @day, @dim, @key, @bytes, @packets)
        ON CONFLICT(server_id, day, dim, key) DO UPDATE SET
          bytes = flow_daily_dims.bytes + excluded.bytes,
          packets = flow_daily_dims.packets + excluded.packets
      `, {
        serverId: Number(serverIdRaw),
        day: dayKey(bucketAt ?? ""),
        dim,
        key,
        bytes: acc.bytes,
        packets: acc.packets,
      })
    }
  }
  minuteRollup.clear()
  minuteDims.clear()
}

async function capDailyAsn(): Promise<void> {
  const today = nowIso().slice(0, 10)
  const rows = await dbAll<{ serverId: number; key: string; bytes: number; packets: number }>(`
    SELECT server_id AS "serverId", key, bytes, packets
    FROM flow_daily_dims
    WHERE day = ? AND dim = 'asn'
    ORDER BY server_id, bytes DESC
  `, [today])
  const byServer = new Map<number, typeof rows>()
  for (const row of rows) {
    const list = byServer.get(row.serverId) ?? []
    list.push(row)
    byServer.set(row.serverId, list)
  }
  for (const [serverId, list] of byServer) {
    if (list.length <= DAILY_ASN_TOP) continue
    let otherBytes = 0
    let otherPackets = 0
    for (const row of list.slice(DAILY_ASN_TOP)) {
      if (row.key === "other") continue
      otherBytes += row.bytes
      otherPackets += row.packets
      await dbQuery(
        `DELETE FROM flow_daily_dims WHERE server_id = ? AND day = ? AND dim = 'asn' AND key = ?`,
        [serverId, today, row.key],
      )
    }
    if (otherBytes > 0) {
      await ensureParentPartition("flow_daily_dims", today)
      await dbQuery(`
        INSERT INTO flow_daily_dims (server_id, day, dim, key, bytes, packets)
        VALUES (?, ?, 'asn', 'other', ?, ?)
        ON CONFLICT(server_id, day, dim, key) DO UPDATE SET
          bytes = flow_daily_dims.bytes + excluded.bytes,
          packets = flow_daily_dims.packets + excluded.packets
      `, [serverId, today, otherBytes, otherPackets])
    }
  }
}

async function pruneStored(): Promise<void> {
  const now = Date.now()
  if (now - lastPruneAt < PRUNE_MS) return
  lastPruneAt = now
  const flowCutoff = new Date(now - retentionHours * 3600_000).toISOString()
  const minuteCutoff = new Date(now - MINUTE_RETENTION_HOURS * 3600_000).toISOString()
  const dailyCutoff = new Date(now - DAILY_RETENTION_DAYS * 86400_000).toISOString().slice(0, 10)
  await dbQuery(`DELETE FROM flow_buckets WHERE bucket_at < ?`, [flowCutoff])
  await dbQuery(`DELETE FROM flow_minute_stats WHERE bucket_at < ?`, [minuteCutoff])
  await dbQuery(`DELETE FROM flow_minute_dims WHERE bucket_at < ?`, [minuteCutoff])
  await dbQuery(`DELETE FROM flow_daily_dims WHERE day < ?`, [dailyCutoff])

  const keep = Math.max(20, topN)
  await dbQuery(`
    DELETE FROM flow_buckets fb
    USING (
      SELECT server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface
      FROM (
        SELECT server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface,
               ROW_NUMBER() OVER (
                 PARTITION BY server_id, bucket_at ORDER BY bytes DESC
               ) AS rn
        FROM flow_buckets
      ) ranked
      WHERE rn > ?
    ) drop_rows
    WHERE fb.server_id = drop_rows.server_id
      AND fb.bucket_at = drop_rows.bucket_at
      AND fb.src = drop_rows.src
      AND fb.dst = drop_rows.dst
      AND fb.proto = drop_rows.proto
      AND fb.src_port = drop_rows.src_port
      AND fb.dst_port = drop_rows.dst_port
      AND fb.in_iface = drop_rows.in_iface
  `, [keep])
  await pruneRipeSqlite(now)
}

function topNPending(rows: PendingFlowRow[]): PendingFlowRow[] {
  const keep = Math.max(20, topN)
  const groups = new Map<string, PendingFlowRow[]>()
  for (const row of rows) {
    const k = `${row.serverId}\0${row.bucketAt}`
    const list = groups.get(k) ?? []
    list.push(row)
    groups.set(k, list)
  }
  const out: PendingFlowRow[] = []
  for (const list of groups.values()) {
    list.sort((a, b) => b.bytes - a.bytes)
    out.push(...list.slice(0, keep))
  }
  return out
}

function persistDue(force: boolean, hasWork: boolean): boolean {
  if (force) return true
  if (!hasWork) return false
  if (minuteBucketIso() !== lastFlushedMinute) return true
  return Date.now() - lastPersistAt >= PERSIST_MS
}

async function upsertFlowBucketsBatch(rows: PendingFlowRow[]): Promise<void> {
  if (rows.length === 0) return
  const days = new Set(rows.map((r) => r.bucketAt))
  for (const bucketAt of days) await ensureParentPartition("flow_buckets", bucketAt)
  await pool.query({
    text: `
      INSERT INTO flow_buckets (
        server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms, nat_src, nat_dst, nat_src_port, nat_dst_port
      )
      SELECT *
      FROM UNNEST(
        $1::bigint[],
        $2::timestamptz[],
        $3::inet[],
        $4::inet[],
        $5::smallint[],
        $6::int[],
        $7::int[],
        $8::bigint[],
        $9::bigint[],
        $10::text[],
        $11::text[],
        $12::inet[],
        $13::bigint[],
        $14::bigint[],
        $15::inet[],
        $16::inet[],
        $17::int[],
        $18::int[]
      ) AS t(server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms, nat_src, nat_dst, nat_src_port, nat_dst_port)
      ON CONFLICT (server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
      DO UPDATE SET
        bytes = flow_buckets.bytes + excluded.bytes,
        packets = flow_buckets.packets + excluded.packets,
        out_iface = CASE WHEN excluded.out_iface != '' THEN excluded.out_iface ELSE flow_buckets.out_iface END,
        next_hop = COALESCE(excluded.next_hop, flow_buckets.next_hop),
        flow_start_ms = CASE
          WHEN excluded.flow_start_ms > 0 AND (flow_buckets.flow_start_ms = 0 OR excluded.flow_start_ms < flow_buckets.flow_start_ms)
          THEN excluded.flow_start_ms ELSE flow_buckets.flow_start_ms END,
        flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms),
        nat_src = COALESCE(excluded.nat_src, flow_buckets.nat_src),
        nat_dst = COALESCE(excluded.nat_dst, flow_buckets.nat_dst),
        nat_src_port = CASE WHEN excluded.nat_src_port > 0 THEN excluded.nat_src_port ELSE flow_buckets.nat_src_port END,
        nat_dst_port = CASE WHEN excluded.nat_dst_port > 0 THEN excluded.nat_dst_port ELSE flow_buckets.nat_dst_port END
    `,
    values: [
      rows.map((r) => r.serverId),
      rows.map((r) => r.bucketAt),
      rows.map((r) => r.src),
      rows.map((r) => r.dst),
      rows.map((r) => r.proto),
      rows.map((r) => r.srcPort),
      rows.map((r) => r.dstPort),
      rows.map((r) => r.bytes),
      rows.map((r) => r.packets),
      rows.map((r) => r.inIface),
      rows.map((r) => r.outIface),
      rows.map((r) => inetOrNull(r.nextHop)),
      rows.map((r) => r.flowStartMs),
      rows.map((r) => r.flowEndMs),
      rows.map((r) => inetOrNull(r.natSrc)),
      rows.map((r) => inetOrNull(r.natDst)),
      rows.map((r) => r.natSrcPort),
      rows.map((r) => r.natDstPort),
    ],
  })
}

const FLOW_UPSERT_SQL = `
    INSERT INTO flow_buckets (
      server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms, nat_src, nat_dst, nat_src_port, nat_dst_port
    ) VALUES (
      @serverId, @bucketAt, @src, @dst, @proto, @srcPort, @dstPort, @bytes, @packets, @inIface, @outIface, @nextHop, @flowStartMs, @flowEndMs, @natSrc, @natDst, @natSrcPort, @natDstPort
    )
    ON CONFLICT(server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
    DO UPDATE SET
      bytes = flow_buckets.bytes + excluded.bytes,
      packets = flow_buckets.packets + excluded.packets,
      out_iface = CASE WHEN excluded.out_iface != '' THEN excluded.out_iface ELSE flow_buckets.out_iface END,
      next_hop = COALESCE(excluded.next_hop, flow_buckets.next_hop),
      flow_start_ms = CASE
        WHEN excluded.flow_start_ms > 0 AND (flow_buckets.flow_start_ms = 0 OR excluded.flow_start_ms < flow_buckets.flow_start_ms)
        THEN excluded.flow_start_ms ELSE flow_buckets.flow_start_ms END,
      flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms),
      nat_src = COALESCE(excluded.nat_src, flow_buckets.nat_src),
      nat_dst = COALESCE(excluded.nat_dst, flow_buckets.nat_dst),
      nat_src_port = CASE WHEN excluded.nat_src_port > 0 THEN excluded.nat_src_port ELSE flow_buckets.nat_src_port END,
      nat_dst_port = CASE WHEN excluded.nat_dst_port > 0 THEN excluded.nat_dst_port ELSE flow_buckets.nat_dst_port END
  `

async function upsertFlowBuckets(rows: PendingFlowRow[]): Promise<number> {
  if (rows.length === 0) return 0
  try {
    await upsertFlowBucketsBatch(rows)
    return rows.length
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err)
    setEngineError(`flow_buckets: ${message}`)
    let stored = 0
    for (const r of rows) {
      try {
        await ensureParentPartition("flow_buckets", r.bucketAt)
        await dbQuery(FLOW_UPSERT_SQL, flowUpsertParams(r))
        stored += 1
      } catch (rowErr) {
        if (stored === 0 && lastError.startsWith("flow_buckets:")) {
          const rowMsg = rowErr instanceof Error ? rowErr.message : String(rowErr)
          setEngineError(`flow_buckets: ${rowMsg}`)
        }
      }
    }
    return stored
  }
}

export async function flushPending(opts?: { force?: boolean; prune?: boolean }): Promise<void> {
  pruneRecent()
  rollFlowRings()
  const force = Boolean(opts?.force)
  const doPrune = opts?.prune !== false
  const hasWork = pending.size > 0 || minuteRollup.size > 0 || minuteDims.size > 0 || factsPendingSize() > 0
  const due = persistDue(force, hasWork)
  try {
    await persistListenerStats(force)
  } catch {
    /* settings row may be absent in unit tests */
  }

  if (!hasWork) {
    if (force && doPrune) {
      try {
        await pruneStored()
      } catch {
        /* prune best-effort */
      }
    }
    lastFlushUsedTransaction = false
    return
  }
  if (!due) {
    lastFlushUsedTransaction = false
    return
  }

  const sanitized: PendingFlowRow[] = []
  let skippedInet = 0
  for (const row of topNPending([...pending.values()].map(toPendingRow))) {
    const clean = sanitizeFlowRow(row)
    if (!clean) {
      skippedInet += 1
      continue
    }
    sanitized.push(clean)
  }
  pending.clear()
  for (const row of sanitized) mergeInto(recent, row)
  if (skippedInet > 0) {
    setEngineError(`flow_buckets: пропуск ${skippedInet} строк с невалидным IP`)
  }

  lastFlushUsedTransaction = false
  lastPersistAt = Date.now()
  lastFlushedMinute = minuteBucketIso()
  try {
    const stored = await upsertFlowBuckets(sanitized)
    lastFlushUsedTransaction = stored === sanitized.length
    rowsStored += stored
    if (stored > 0) bumpDataEpoch()
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err)
    setEngineError(`flow_buckets: ${message}`)
  }
  try {
    await upsertMinuteAndDaily()
    await capDailyAsn()
  } catch {
    /* rollup best-effort */
  }
  try {
    await flushFlowFacts()
  } catch {
    /* statistics cube best-effort */
  }
  if (doPrune) {
    try {
      await pruneStored()
    } catch {
      /* prune best-effort */
    }
  }
}

export function lastFlushUsedTransactionForTests(): boolean {
  return lastFlushUsedTransaction
}

export async function flushEngineNow(opts?: { prune?: boolean }): Promise<void> {
  await flushPending({ force: true, prune: opts?.prune })
}

export async function flushPendingForTests(): Promise<void> {
  await flushEngineNow()
}

export function onEngineTick(): void {
  void flushPending()
}

export function ingestParsedFlowsForServerForTests(serverId: number, flows: ParsedFlowInput[]): void {
  queueParsedFlows(serverId, flows)
  rollFlowRings()
}

export function resetEngineForTests(): void {
  pending.clear()
  recent.clear()
  tickAccum.clear()
  rings.clear()
  minuteRollup.clear()
  minuteDims.clear()
  resetFactsForTests()
  packetsReceived = 0
  lastExporterIp = null
  lastError = ""
  lastDatagramAt = null
  dropped = 0
  rowsStored = 0
  lastFlushUsedTransaction = false
  lastPruneAt = 0
  lastPassiveCheckpointAt = Date.now()
  lastPersistAt = 0
  lastFlushedMinute = ""
  lastStatsPersistAt = 0
  lastPersistedStats = null
  bumpDataEpoch()
  pendingCap = MAX_PENDING
}

export function pendingSizeForTests(): number {
  return pending.size
}

export function droppedForTests(): number {
  return dropped
}

/** Снимок минутных dims (dim → key → bytes) для тестов обогащения потоков. */
export function minuteDimsSnapshotForTests(): Map<string, Map<string, { bytes: number; packets: number }>> {
  const out = new Map<string, Map<string, { bytes: number; packets: number }>>()
  for (const [k, acc] of minuteDims) {
    // dimKey: serverId\0bucketAt\0dim\0key
    const parts = k.split("\0")
    const dim = parts[2] ?? ""
    const key = parts.slice(3).join("\0")
    let byKey = out.get(dim)
    if (!byKey) {
      byKey = new Map()
      out.set(dim, byKey)
    }
    const prev = byKey.get(key)
    byKey.set(key, {
      bytes: (prev?.bytes ?? 0) + acc.bytes,
      packets: (prev?.packets ?? 0) + acc.packets,
    })
  }
  return out
}