import { Worker } from "node:worker_threads"
import { existsSync, statSync } from "node:fs"
import path from "node:path"
import { gte, sql } from "drizzle-orm"
import { beginSqliteExclusiveOp, db, endSqliteExclusiveOp, sqliteDatabase } from "../db/index.js"
import { env } from "../config.js"
import { flowBuckets, servers } from "../db/schema.js"
import type { FlowPurgeDto, FlowStatsDto, FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
import { protoName, type ParsedFlowInput } from "./traffic-flow-parse.js"
import type { CollectorHeartbeat, ExporterMapPayload, MainToWorker, WorkerToMain } from "./traffic-flow-collector-ipc.js"
import {
  attachEngineSqlite,
  applyRingSnapshot,
  configureEngine,
  flushPending,
  getEngineStats,
  getRingMbps as engineGetRingMbps,
  ingestParsedFlowsForServerForTests as engineIngestForServer,
  lastFlushUsedTransactionForTests as engineLastFlushTx,
  listLiveFlowRows as engineListLive,
  peekPendingFlows,
  queueParsedFlows,
  resetEngineForTests,
  resolveServerId,
  rollFlowRings,
  setExporterResolveCtx,
  type PendingFlowRow,
} from "./traffic-flow-engine.js"
import {
  getTrafficFlowSettingsRow,
  listHostPeers,
  resetFlowIngestCounters,
} from "./traffic-flow-settings.js"
import { applicationName } from "./traffic-flow-apps.js"
import { resolveIfaceName } from "./traffic-flow-ifaces.js"

export type { PendingFlowRow }

export interface FlowListenerState {
  bound: boolean
  address: string | null
}

export interface FlowWorkerHealth {
  alive: boolean
  bound: boolean
  pendingSize: number
  dropped: number
  packetsReceived: number
}

let worker: Worker | null = null
let restartTimer: ReturnType<typeof setTimeout> | null = null
let restartAttempts = 0
let lastHeartbeat: CollectorHeartbeat | null = null
let state: FlowListenerState = { bound: false, address: null }
let wantListen = false

attachEngineSqlite(sqliteDatabase)

function workerFileUrl(): URL {
  const ts = import.meta.url.includes(".ts")
  return new URL(
    ts ? "./traffic-flow-collector-worker.ts" : "./traffic-flow-collector-worker.js",
    import.meta.url,
  )
}

export function buildExporterMapPayload(): ExporterMapPayload {
  const settings = getTrafficFlowSettingsRow()
  const rows = db.select({
    id: servers.id,
    host: servers.host,
    mgmtTunnelIp: servers.mgmtTunnelIp,
  }).from(servers).all()
  const byTunnelIp: Array<[string, number]> = []
  const hostIps: Array<[string, number]> = []
  for (const row of rows) {
    if (row.mgmtTunnelIp) byTunnelIp.push([row.mgmtTunnelIp, row.id])
    if (/^\d{1,3}(?:\.\d{1,3}){3}$/.test(row.host)) hostIps.push([row.host, row.id])
  }
  return {
    overlayPrefix: settings.prefix,
    byTunnelIp,
    peers: listHostPeers(),
    hostIps,
  }
}

function applyExporterCtxFromDb(): void {
  const payload = buildExporterMapPayload()
  setExporterResolveCtx({
    overlayPrefix: payload.overlayPrefix,
    byTunnelIp: new Map(payload.byTunnelIp),
    peers: payload.peers,
    hostIps: new Map(payload.hostIps),
  })
}

function postToWorker(msg: MainToWorker): void {
  worker?.postMessage(msg)
}

function handleWorkerMessage(msg: WorkerToMain): void {
  if (msg.type === "heartbeat") {
    lastHeartbeat = msg.payload
    state = { bound: msg.payload.bound, address: msg.payload.address }
    applyRingSnapshot(msg.payload.rings)
    restartAttempts = 0
    return
  }
  if (msg.type === "error") {
    lastHeartbeat = lastHeartbeat
      ? { ...lastHeartbeat, lastError: msg.payload.message, workerAlive: true }
      : null
  }
}

function spawnWorker(): void {
  stopWorkerProcess()
  const settings = getTrafficFlowSettingsRow()
  configureEngine({ topN: settings.topN, retentionHours: settings.retentionHours })
  applyExporterCtxFromDb()
  const w = new Worker(workerFileUrl(), { execArgv: process.execArgv })
  w.on("message", (msg: WorkerToMain) => handleWorkerMessage(msg))
  w.on("error", (err) => {
    state = { bound: false, address: null }
    lastHeartbeat = lastHeartbeat
      ? { ...lastHeartbeat, workerAlive: false, lastError: err.message, bound: false }
      : {
          bound: false,
          address: null,
          packetsReceived: 0,
          lastExporterIp: null,
          lastError: err.message,
          lastDatagramAt: null,
          pendingSize: 0,
          dropped: 0,
          rowsStored: 0,
          workerAlive: false,
          rings: [],
        }
  })
  w.on("exit", (code) => {
    worker = null
    state = { bound: false, address: null }
    if (!wantListen) return
    const delay = Math.min(30_000, 1000 * 2 ** restartAttempts)
    restartAttempts += 1
    restartTimer = setTimeout(() => {
      if (wantListen) spawnWorker()
    }, delay)
    void code
  })
  worker = w
  const host = process.env.FLOW_LISTEN_HOST?.trim() || settings.collectorIp || "127.0.0.1"
  postToWorker({
    type: "start",
    payload: {
      dbPath: env.DATABASE_PATH,
      listenHost: host,
      listenPort: settings.flowListenPort,
      topN: settings.topN,
      retentionHours: settings.retentionHours,
      exporterMap: buildExporterMapPayload(),
    },
  })
}

function stopWorkerProcess(): void {
  if (restartTimer) {
    clearTimeout(restartTimer)
    restartTimer = null
  }
  if (worker) {
    try {
      postToWorker({ type: "stop" })
      void worker.terminate()
    } catch {
      /* ignore */
    }
    worker = null
  }
}

export function reattachFlowSqlite(): void {
  attachEngineSqlite(sqliteDatabase)
}

export function applyHeartbeatForTests(payload: CollectorHeartbeat): void {
  handleWorkerMessage({ type: "heartbeat", payload })
}

export function simulateWorkerExitForTests(): number {
  worker = null
  state = { bound: false, address: null }
  lastHeartbeat = lastHeartbeat ? { ...lastHeartbeat, workerAlive: false, bound: false } : null
  if (!wantListen) return restartAttempts
  restartAttempts += 1
  return restartAttempts
}

export function setWantListenForTests(value: boolean): void {
  wantListen = value
}

export function getFlowListenerState(): FlowListenerState {
  return state
}

export function getFlowWorkerHealth(): FlowWorkerHealth {
  const hb = lastHeartbeat
  const mem = getEngineStats()
  return {
    alive: Boolean(worker) && (hb?.workerAlive ?? false),
    bound: state.bound,
    pendingSize: hb?.pendingSize ?? mem.pendingSize,
    dropped: hb?.dropped ?? mem.dropped,
    packetsReceived: hb?.packetsReceived ?? mem.packetsReceived,
  }
}

export function getFlowRuntimeCounters() {
  const settings = getTrafficFlowSettingsRow()
  const hb = lastHeartbeat
  return {
    packetsReceived: hb?.packetsReceived ?? settings.packetsReceived,
    lastExporterIp: hb?.lastExporterIp ?? settings.lastExporterIp ?? null,
    lastError: (hb?.lastError ?? settings.lastError) || null,
    lastDatagramAt: hb?.lastDatagramAt ?? settings.lastDatagramAt ?? null,
    dropped: hb?.dropped ?? 0,
  }
}

export function startTrafficFlowListener() {
  stopTrafficFlowListener()
  const settings = getTrafficFlowSettingsRow()
  if (!settings.enabled) {
    wantListen = false
    state = { bound: false, address: null }
    return
  }
  wantListen = true
  spawnWorker()
}

export function stopTrafficFlowListener() {
  wantListen = false
  stopWorkerProcess()
  try {
    flushPending()
  } catch {
    /* ignore */
  }
  state = { bound: false, address: null }
}

export function refreshFlowExporterMap(): void {
  applyExporterCtxFromDb()
  postToWorker({ type: "updateExporterMap", payload: buildExporterMapPayload() })
}

export function getRingMbps(serverId: number, iface = "__all__") {
  return engineGetRingMbps(serverId, iface)
}

function mergeInto(map: Map<string, PendingFlowRow>, row: PendingFlowRow): void {
  const key = `${row.serverId}|${row.bucketAt}|${row.src}|${row.dst}|${row.proto}|${row.srcPort}|${row.dstPort}|${row.inIface}`
  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 })
}

export function listLiveFlowRows(sinceIso: string): PendingFlowRow[] {
  if (worker && lastHeartbeat?.workerAlive) {
    return listStoredFlowRows(sinceIso)
  }
  return engineListLive(sinceIso)
}

export function listStoredFlowRows(sinceIso: string): PendingFlowRow[] {
  const settings = getTrafficFlowSettingsRow()
  const cap = Math.max(20, settings.topN) * 60
  const stored = db.select().from(flowBuckets)
    .where(gte(flowBuckets.bucketAt, sinceIso))
    .orderBy(sql`${flowBuckets.bytes} DESC`)
    .limit(cap)
    .all()
  const merged = new Map<string, PendingFlowRow>()
  for (const r of stored) {
    mergeInto(merged, {
      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 ?? 0,
      flowEndMs: r.flowEndMs ?? 0,
    })
  }
  if (!worker) {
    for (const p of peekPendingFlows()) {
      if (p.bucketAt < sinceIso) continue
      mergeInto(merged, p)
    }
  }
  return [...merged.values()]
}

export function listFlowRowsForWindow(minutes: number): PendingFlowRow[] {
  const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString()
  if (minutes <= 15 && !worker) return listLiveFlowRows(sinceIso)
  return listStoredFlowRows(sinceIso)
}

export function listFlowTalkers(minutes = 5): FlowStatsDto {
  const settings = getTrafficFlowSettingsRow()
  const runtime = getFlowRuntimeCounters()
  const rows = listFlowRowsForWindow(minutes)
  const serverRows = db.select().from(servers).all()
  const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host]))
  const agg = new Map<string, FlowTalkerDto & { rawBytes: number }>()
  const protoBytes = new Map<number, number>()
  const srcs = new Set<string>()
  const dsts = new Set<string>()
  const exporters = new Set<number>()
  let totalBytes = 0
  for (const r of rows) {
    const resolved = resolveIfaceName(r.serverId, r.inIface)
    const key = `${r.serverId}|${r.src}|${r.dst}|${r.proto}|${r.srcPort}|${r.dstPort}|${r.inIface}`
    const prev = agg.get(key)
    const bytes = r.bytes
    totalBytes += bytes
    srcs.add(r.src)
    dsts.add(r.dst)
    exporters.add(r.serverId)
    protoBytes.set(r.proto, (protoBytes.get(r.proto) ?? 0) + bytes)
    if (prev) {
      prev.rawBytes += bytes
      prev.bytes += bytes
      prev.packets += r.packets
    } else {
      agg.set(key, {
        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,
        packets: r.packets,
        bps: 0,
        inIface: resolved.name,
        inIfaceIndex: resolved.index,
        application: applicationName(r.proto, r.dstPort, r.srcPort),
        rawBytes: bytes,
      })
    }
  }
  const windowSec = Math.max(60, minutes * 60)
  const talkers = [...agg.values()]
    .map((t) => ({ ...t, bps: (t.rawBytes * 8) / windowSec }))
    .sort((a, b) => b.bytes - a.bytes)
    .slice(0, settings.topN)
    .map(({ rawBytes: _raw, ...rest }) => rest)
  let topProto = "—"
  let topProtoBytes = 0
  for (const [p, b] of protoBytes) {
    if (b > topProtoBytes) {
      topProtoBytes = b
      topProto = protoName(p)
    }
  }
  return {
    exportersOnline: exporters.size,
    bytesPerMin: minutes > 0 ? totalBytes / minutes : totalBytes,
    uniqueSrc: srcs.size,
    uniqueDst: dsts.size,
    topProto,
    talkers,
    lastExporterIp: runtime.lastExporterIp,
    lastError: runtime.lastError,
    packetsReceived: runtime.packetsReceived,
    lastDatagramAt: runtime.lastDatagramAt,
    listenerBound: state.bound,
    listenerAddress: state.address,
  }
}

export function ingestParsedFlowsForTests(exporterIp: string, flows: ParsedFlowInput[]) {
  applyExporterCtxFromDb()
  const serverId = resolveServerId(exporterIp)
  if (serverId == null) return
  queueParsedFlows(serverId, flows)
  rollFlowRings()
  flushPending()
}

export function ingestParsedFlowsForServerForTests(serverId: number, flows: ParsedFlowInput[]) {
  engineIngestForServer(serverId, flows)
}

export function resetFlowRingsForTests() {
  resetEngineForTests()
  attachEngineSqlite(sqliteDatabase)
  lastHeartbeat = null
  wantListen = false
  restartAttempts = 0
}

export function lastFlushUsedTransactionForTests(): boolean {
  return engineLastFlushTx()
}

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

function tableCount(name: string): number {
  const row = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM ${name}`).get() as { n: number }
  return Number(row?.n) || 0
}

function dbFileBytes(): number {
  const resolved = path.resolve(process.cwd(), env.DATABASE_PATH)
  if (!existsSync(resolved)) return 0
  return statSync(resolved).size
}

async function stopWorkerProcessAsync(): Promise<void> {
  if (restartTimer) {
    clearTimeout(restartTimer)
    restartTimer = null
  }
  if (!worker) return
  const current = worker
  worker = null
  try {
    current.postMessage({ type: "stop" })
    await current.terminate()
  } catch {
    /* ignore */
  }
}

/** Удаляет сессии, minute/daily rollup и сжимает SQLite. Ключи WG и пиры JH не трогает. */
export async function purgeTrafficFlowStore(): Promise<FlowPurgeDto> {
  beginSqliteExclusiveOp()
  try {
    wantListen = false
    await stopWorkerProcessAsync()
    resetEngineForTests()
    attachEngineSqlite(sqliteDatabase)
    lastHeartbeat = null
    state = { bound: false, address: null }
    const fileBytesBefore = dbFileBytes()
    const deleted = {
      buckets: tableCount("flow_buckets"),
      minuteStats: tableCount("flow_minute_stats"),
      minuteDims: tableCount("flow_minute_dims"),
      dailyDims: tableCount("flow_daily_dims"),
    }
    sqliteDatabase.exec(`
      DELETE FROM flow_buckets;
      DELETE FROM flow_minute_stats;
      DELETE FROM flow_minute_dims;
      DELETE FROM flow_daily_dims;
    `)
    resetFlowIngestCounters()
    try {
      sqliteDatabase.pragma("wal_checkpoint(TRUNCATE)")
    } catch {
      /* ignore */
    }
    let vacuumed = false
    try {
      sqliteDatabase.exec("VACUUM")
      vacuumed = true
    } catch {
      vacuumed = false
    }
    return {
      ok: true,
      deleted,
      fileBytesBefore,
      fileBytesAfter: dbFileBytes(),
      vacuumed,
    }
  } finally {
    try {
      startTrafficFlowListener()
    } catch {
      /* ingest мог остаться выключенным */
    }
    endSqliteExclusiveOp()
  }
}

export { peekPendingFlows }
export { setPendingCapForTests } from "./traffic-flow-engine.js"
export { maybeRefreshIfaces, setRefreshIfacesForTests } from "./traffic-flow-ifaces.js"
