import { createSocket, type Socket } from "node:dgram"
import { parentPort } from "node:worker_threads"
import { sqliteDatabase } from "../db/index.js"
import type {
  CollectorStartPayload,
  ExporterMapPayload,
  MainToWorker,
  WorkerToMain,
} from "./traffic-flow-collector-ipc.js"
import {
  TICK_MS,
  attachEngineSqlite,
  configureEngine,
  flushPending,
  getEngineStats,
  ingestDatagram,
  setEngineError,
  setExporterResolveCtx,
  snapshotRings,
} from "./traffic-flow-engine.js"

let socket: Socket | null = null
let flushTimer: ReturnType<typeof setInterval> | null = null
let bound = false
let address: string | null = null
let attached = false

function send(msg: WorkerToMain): void {
  parentPort?.postMessage(msg)
}

function heartbeat(): void {
  const stats = getEngineStats()
  send({
    type: "heartbeat",
    payload: {
      bound,
      address,
      packetsReceived: stats.packetsReceived,
      lastExporterIp: stats.lastExporterIp,
      lastError: stats.lastError,
      lastDatagramAt: stats.lastDatagramAt,
      pendingSize: stats.pendingSize,
      dropped: stats.dropped,
      rowsStored: stats.rowsStored,
      workerAlive: true,
      rings: snapshotRings(),
    },
  })
}

function applyExporterMap(payload: ExporterMapPayload): void {
  setExporterResolveCtx({
    overlayPrefix: payload.overlayPrefix,
    byTunnelIp: new Map(payload.byTunnelIp),
    peers: payload.peers,
    hostIps: new Map(payload.hostIps),
  })
}

function ensureSqlite(): void {
  if (attached) return
  attachEngineSqlite(sqliteDatabase)
  attached = true
}

function stopListener(): void {
  if (flushTimer) {
    clearInterval(flushTimer)
    flushTimer = null
  }
  try {
    flushPending()
  } catch (e) {
    setEngineError(e instanceof Error ? e.message : String(e))
  }
  if (socket) {
    try { socket.close() } catch { /* ignore */ }
    socket = null
  }
  bound = false
  address = null
}

function startListener(payload: CollectorStartPayload): void {
  stopListener()
  ensureSqlite()
  configureEngine({ topN: payload.topN, retentionHours: payload.retentionHours })
  applyExporterMap(payload.exporterMap)

  const sock = createSocket("udp4")
  sock.on("error", (err) => {
    setEngineError(err.message)
    bound = false
    address = null
    send({ type: "error", payload: { message: err.message } })
    heartbeat()
  })
  sock.on("message", (msg, rinfo) => {
    try {
      ingestDatagram(msg, rinfo.address)
    } catch (e) {
      setEngineError(e instanceof Error ? e.message : String(e))
    }
  })
  try {
    sock.setRecvBufferSize(8 * 1024 * 1024)
  } catch {
    /* platform may ignore */
  }
  sock.bind(payload.listenPort, payload.listenHost, () => {
    bound = true
    address = `${payload.listenHost}:${payload.listenPort}`
    setEngineError("")
    heartbeat()
  })
  socket = sock
  flushTimer = setInterval(() => {
    try {
      flushPending()
    } catch (e) {
      setEngineError(e instanceof Error ? e.message : String(e))
    }
    heartbeat()
  }, TICK_MS)
}

parentPort?.on("message", (msg: MainToWorker) => {
  try {
    if (msg.type === "start") startListener(msg.payload)
    else if (msg.type === "stop") {
      stopListener()
      heartbeat()
    } else if (msg.type === "updateExporterMap") applyExporterMap(msg.payload)
    else if (msg.type === "updateSettings") configureEngine(msg.payload)
  } catch (e) {
    const message = e instanceof Error ? e.message : String(e)
    setEngineError(message)
    send({ type: "error", payload: { message } })
  }
})
