import type Database from "better-sqlite3"
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 { classifyFlowDst } from "./traffic-flow-classify.js"
import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js"
import { isIsoCountry } from "./traffic-flow-brands.js"
import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js"

type SqliteHandle = InstanceType<typeof Database>

export const TICK_MS = 2_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

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
}

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 sqliteRef: SqliteHandle | null = null
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 exporterCtx: ExporterResolveCtx | null = null

const PRUNE_MS = 5 * 60_000
const LIVE_WINDOW_MS = 15 * 60_000

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: SqliteHandle): void {
  sqliteRef = handle
}

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 {
  const bucketAt = minuteBucketIso()
  const ripeMisses: string[] = []
  for (const raw of flows) {
    const flow = normalizeParsedFlow(raw)
    addToTick(serverId, flow, flow.bytes)
    bumpRollup(serverId, bucketAt, flow, flow.bytes, flow.packets)
    const ripe = lookupRipeCached(flow.dst)
    if (flow.dst && !ripe) ripeMisses.push(flow.dst)
    const classified = classifyFlowDst(flow.dst, flow.proto, flow.dstPort, flow.srcPort, ripe)
    const app = applicationName(flow.proto, flow.dstPort, flow.srcPort)
    const country = ripe?.ok && isIsoCountry(ripe.country)
      ? ripe.country
      : (ripe?.ok ? "" : "unknown")
    const asnKey = ripe?.ok && ripe.asn ? String(ripe.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)

    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.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,
  }
}

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.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 })
  }
}

function persistListenerStats(handle: SqliteHandle): void {
  handle.prepare(`
    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
  `).run({
    packetsReceived,
    lastDatagramAt,
    lastExporterIp,
    lastError,
    updatedAt: nowIso(),
  })
}

function upsertMinuteAndDaily(handle: SqliteHandle): void {
  const upsertMinute = handle.prepare(`
    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 = bytes + excluded.bytes,
      packets = packets + excluded.packets,
      unique_src = MAX(unique_src, excluded.unique_src),
      unique_dst = MAX(unique_dst, excluded.unique_dst),
      conversations = conversations + excluded.conversations
  `)
  const upsertDim = handle.prepare(`
    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 = bytes + excluded.bytes,
      packets = packets + excluded.packets
  `)
  const upsertDaily = handle.prepare(`
    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 = bytes + excluded.bytes,
      packets = packets + excluded.packets
  `)

  const tx = handle.transaction(() => {
    for (const [k, acc] of minuteRollup) {
      const [serverIdRaw, bucketAt] = k.split("\0")
      upsertMinute.run({
        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")
      upsertDim.run({
        serverId: Number(serverIdRaw),
        bucketAt,
        dim,
        key,
        bytes: acc.bytes,
        packets: acc.packets,
      })
      if (dim === "country" || dim === "service" || dim === "asn") {
        upsertDaily.run({
          serverId: Number(serverIdRaw),
          day: dayKey(bucketAt ?? ""),
          dim,
          key,
          bytes: acc.bytes,
          packets: acc.packets,
        })
      }
    }
  })
  tx()
  minuteRollup.clear()
  minuteDims.clear()
}

function capDailyAsn(handle: SqliteHandle): void {
  const today = nowIso().slice(0, 10)
  const rows = handle.prepare(`
    SELECT server_id AS serverId, key, bytes, packets
    FROM flow_daily_dims
    WHERE day = ? AND dim = 'asn'
    ORDER BY server_id, bytes DESC
  `).all(today) as Array<{ serverId: number; key: string; bytes: number; packets: number }>
  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)
  }
  const del = handle.prepare(`
    DELETE FROM flow_daily_dims WHERE server_id = ? AND day = ? AND dim = 'asn' AND key = ?
  `)
  const upsertOther = handle.prepare(`
    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 = bytes + excluded.bytes,
      packets = packets + excluded.packets
  `)
  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
      del.run(serverId, today, row.key)
    }
    if (otherBytes > 0) upsertOther.run(serverId, today, otherBytes, otherPackets)
  }
}

function pruneStored(handle: SqliteHandle): 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)
  handle.prepare(`DELETE FROM flow_buckets WHERE bucket_at < ?`).run(flowCutoff)
  handle.prepare(`DELETE FROM flow_minute_stats WHERE bucket_at < ?`).run(minuteCutoff)
  handle.prepare(`DELETE FROM flow_minute_dims WHERE bucket_at < ?`).run(minuteCutoff)
  handle.prepare(`DELETE FROM flow_daily_dims WHERE day < ?`).run(dailyCutoff)

  const keep = Math.max(20, topN)
  try {
    handle.prepare(`
      DELETE FROM flow_buckets WHERE id IN (
        SELECT id FROM (
          SELECT id, ROW_NUMBER() OVER (
            PARTITION BY server_id, bucket_at ORDER BY bytes DESC
          ) AS rn
          FROM flow_buckets
        ) ranked WHERE rn > ?
      )
    `).run(keep)
  } catch {
    const buckets = handle.prepare(`
      SELECT DISTINCT server_id AS serverId, bucket_at AS bucketAt FROM flow_buckets
    `).all() as Array<{ serverId: number; bucketAt: string }>
    for (const b of buckets) {
      const rows = handle.prepare(`
        SELECT id, bytes FROM flow_buckets
        WHERE server_id = ? AND bucket_at = ?
        ORDER BY bytes DESC
      `).all(b.serverId, b.bucketAt) as Array<{ id: number; bytes: number }>
      for (const extra of rows.slice(keep)) {
        handle.prepare(`DELETE FROM flow_buckets WHERE id = ?`).run(extra.id)
      }
    }
  }
}

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
}

export function flushPending(): void {
  pruneRecent()
  rollFlowRings()
  const handle = sqliteRef
  if (!handle) {
    lastFlushUsedTransaction = false
    return
  }
  persistListenerStats(handle)
  if (pending.size === 0 && minuteRollup.size === 0 && minuteDims.size === 0) {
    pruneStored(handle)
    lastFlushUsedTransaction = false
    return
  }
  const rows = topNPending([...pending.values()].map(toPendingRow))
  pending.clear()
  for (const row of rows) mergeInto(recent, row)

  const upsertFlow = handle.prepare(`
    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
    ) VALUES (
      @serverId, @bucketAt, @src, @dst, @proto, @srcPort, @dstPort, @bytes, @packets, @inIface, @outIface, @nextHop, @flowStartMs, @flowEndMs
    )
    ON CONFLICT(server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface)
    DO UPDATE SET
      bytes = bytes + excluded.bytes,
      packets = packets + excluded.packets,
      out_iface = CASE WHEN excluded.out_iface != '' THEN excluded.out_iface ELSE out_iface END,
      next_hop = CASE WHEN excluded.next_hop != '' THEN excluded.next_hop ELSE next_hop END,
      flow_start_ms = CASE
        WHEN excluded.flow_start_ms > 0 AND (flow_start_ms = 0 OR excluded.flow_start_ms < flow_start_ms)
        THEN excluded.flow_start_ms ELSE flow_start_ms END,
      flow_end_ms = MAX(flow_end_ms, excluded.flow_end_ms)
  `)
  lastFlushUsedTransaction = false
  try {
    const tx = handle.transaction((batch: PendingFlowRow[]) => {
      for (const r of batch) {
        upsertFlow.run({
          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: r.nextHop,
          flowStartMs: r.flowStartMs,
          flowEndMs: r.flowEndMs,
        })
      }
    })
    tx(rows)
    lastFlushUsedTransaction = true
    rowsStored += rows.length
  } catch {
    for (const r of rows) {
      try {
        upsertFlow.run({
          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: r.nextHop,
          flowStartMs: r.flowStartMs,
          flowEndMs: r.flowEndMs,
        })
        rowsStored += 1
      } catch {
        /* ignore single-row failures */
      }
    }
  }
  try {
    upsertMinuteAndDaily(handle)
    capDailyAsn(handle)
  } catch {
    /* rollup best-effort */
  }
  pruneStored(handle)
  try {
    handle.pragma("wal_checkpoint(TRUNCATE)")
  } catch {
    /* ignore */
  }
}

export function lastFlushUsedTransactionForTests(): boolean {
  return lastFlushUsedTransaction
}

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

export function onEngineTick(): 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()
  packetsReceived = 0
  lastExporterIp = null
  lastError = ""
  lastDatagramAt = null
  dropped = 0
  rowsStored = 0
  lastFlushUsedTransaction = false
  lastPruneAt = 0
  pendingCap = MAX_PENDING
}

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

export function droppedForTests(): number {
  return dropped
}
