/**
 * Opt-in process top-N collector (Phase 3).
 *
 * Enable: PEARDATA_PROCESSES=1
 * Limit:  PEARDATA_PROCESSES_TOP=8 (default)
 *
 * Emits:
 *   processes.top_cpu  — % of one host CPU for top processes (by name)
 *   processes.top_rss  — RSS MiB for top processes (by RSS)
 *   processes.top_io   — read+write byte rates for top processes (by I/O)
 *   processes.top_threads — thread count for top processes (by threads)
 *
 * Linux /proc only; no-op elsewhere.
 */
import fs from 'fs'
import path from 'path'
import os from 'os'
import { EventEmitter } from 'events'
import { SAMPLE_INTERVAL_MS, registerChart } from '../../../shared/metrics.js'
import { readFileBuf } from '../../utils/fd-cache.js'
import logger from '../../utils/logger.js'

const log = logger.child('processes')

export function isProcessCollectorEnabled() {
  const v = process.env.PEARDATA_PROCESSES
  if (v === '0' || v === 'off' || v === 'false') return false
  if (v === '1' || v === 'on' || v === 'true') return true
  // Default on for Linux — powers Charts top-* series alongside the Processes tab
  return os.platform() === 'linux'
}

const HOST_CPUS = (() => {
  try {
    return os.cpus().length || 1
  } catch {
    return 1
  }
})()

const PROC_LIMIT = (() => {
  const n = Number(process.env.PEARDATA_PROCESSES_TOP)
  return Number.isFinite(n) && n > 0 ? Math.min(32, Math.floor(n)) : 8
})()

function sanitizeDim(name) {
  const s = String(name || 'unknown')
    .replace(/[^\w.+-]/g, '_')
    .replace(/^_+|_+$/g, '')
    .slice(0, 48)
  return s || 'unknown'
}

const PROC_EXTRA_CACHE = new Map()
const PROC_EXTRA_TICK = 5
let procTickCount = 0

/**
 * @returns {Array<{ pid: number, name: string, utime: number, stime: number, rssPages: number, threads: number, readBytes: number|null, writeBytes: number|null, ioSampleTs: number|null }>|null}
 */
export function listProcStats() {
  if (os.platform() !== 'linux') return null
  let dirs
  try {
    dirs = fs.readdirSync('/proc')
  } catch {
    return null
  }
  procTickCount = (procTickCount + 1) % PROC_EXTRA_TICK
  const readExtra = procTickCount === 0
  if (readExtra) PROC_EXTRA_CACHE.clear()
  const now = Date.now()

  /** @type {Array<{ pid: number, name: string, utime: number, stime: number, rssPages: number, threads: number, readBytes: number|null, writeBytes: number|null, ioSampleTs: number|null }>} */
  const out = []
  for (const ent of dirs) {
    if (!/^\d+$/.test(ent)) continue
    const pid = Number(ent)
    const raw = readFileBuf(path.join('/proc', ent, 'stat'))
    if (!raw) continue
    const open = raw.indexOf('(')
    const close = raw.lastIndexOf(')')
    if (open < 0 || close < open) continue
    const name = raw.slice(open + 1, close)
    const rest = raw.slice(close + 2).split(/\s+/)
    const utime = Number(rest[11])
    const stime = Number(rest[12])
    const rssPages = Number(rest[21])
    if (!Number.isFinite(utime) || !Number.isFinite(stime)) continue

    let threads = 0
    let readBytes = null
    let writeBytes = null
    let ioSampleTs = null

    if (readExtra) {
      const status = readFileBuf(path.join('/proc', ent, 'status'))
      if (status) {
        const m = status.match(/^Threads:\s*(\d+)/m)
        if (m) threads = Number(m[1]) || 0
      }
      const ioRaw = readFileBuf(path.join('/proc', ent, 'io'))
      if (ioRaw) {
        const rb = ioRaw.match(/^read_bytes:\s*(\d+)/m)
        const wb = ioRaw.match(/^write_bytes:\s*(\d+)/m)
        if (rb) readBytes = Number(rb[1]) || 0
        if (wb) writeBytes = Number(wb[1]) || 0
        if (readBytes != null || writeBytes != null) ioSampleTs = now
      }
      PROC_EXTRA_CACHE.set(pid, { threads, readBytes, writeBytes, ioSampleTs })
    } else {
      const cached = PROC_EXTRA_CACHE.get(pid)
      if (cached) {
        threads = cached.threads
        readBytes = cached.readBytes
        writeBytes = cached.writeBytes
        ioSampleTs = cached.ioSampleTs
      }
    }

    out.push({
      pid,
      name: sanitizeDim(name),
      utime,
      stime,
      rssPages: Number.isFinite(rssPages) ? rssPages : 0,
      threads,
      readBytes,
      writeBytes,
      ioSampleTs,
    })
  }
  return out
}

function pageSize() {
  try {
    return os.constants?.os?.PAGE_SIZE || 4096
  } catch {
    return 4096
  }
}

/**
 * @param {string[]} names
 * @param {'cpu'|'rss'|'io'|'threads'} kind
 */
function registerTopChart(names, kind) {
  const dims = [...new Set(names)].slice(0, PROC_LIMIT).map((id) => ({
    id,
    name: id,
    algorithm: 'absolute',
  }))
  if (!dims.length) {
    dims.push({ id: '_idle', name: '_idle', algorithm: 'absolute' })
  }
  /** @type {import('../../../shared/metrics.js').ChartDef} */
  let def
  if (kind === 'cpu') {
    def = {
      id: 'processes.top_cpu',
      name: 'processes.top_cpu',
      context: 'processes.top_cpu',
      title: 'Top processes CPU',
      units: 'percentage',
      family: 'processes',
      chartType: 'stacked',
      priority: 6000,
      plugin: 'processes',
      dimensions: dims,
    }
  } else if (kind === 'rss') {
    def = {
      id: 'processes.top_rss',
      name: 'processes.top_rss',
      context: 'processes.top_rss',
      title: 'Top processes RSS',
      units: 'MiB',
      family: 'processes',
      chartType: 'stacked',
      priority: 6010,
      plugin: 'processes',
      dimensions: dims,
    }
  } else if (kind === 'io') {
    def = {
      id: 'processes.top_io',
      name: 'processes.top_io',
      context: 'processes.top_io',
      title: 'Top processes I/O',
      units: 'KiB/s',
      family: 'processes',
      chartType: 'stacked',
      priority: 6020,
      plugin: 'processes',
      dimensions: dims,
    }
  } else {
    def = {
      id: 'processes.top_threads',
      name: 'processes.top_threads',
      context: 'processes.top_threads',
      title: 'Top processes threads',
      units: 'threads',
      family: 'processes',
      chartType: 'stacked',
      priority: 6030,
      plugin: 'processes',
      dimensions: dims,
    }
  }
  registerChart(def)
  return def
}

export class ProcessCollector extends EventEmitter {
  constructor(opts = {}) {
    super()
    this.intervalMs = opts.intervalMs || (() => {
      const p = process.env.PEARDATA_PROCESSES_MS
      if (p && Number.isFinite(Number(p)) && Number(p) > 0) return Number(p)
      return 3000
    })()
    this._timer = null
    /** @type {Map<number, { ticks: number, wallMs: number, name: string, readBytes: number, writeBytes: number }>|null} */
    this._prev = null
    this._pageBytes = 4096
  }

  start() {
    if (this._timer) return
    this._pageBytes = pageSize()
    log.info('Process top-N collector started', { top: PROC_LIMIT })
    this._tick()
    this._timer = setInterval(() => this._tick(), this.intervalMs)
    if (typeof this._timer.unref === 'function') this._timer.unref()
  }

  stop() {
    if (this._timer) {
      clearInterval(this._timer)
      this._timer = null
    }
  }

  _tick() {
    const t0 = performance.now()
    try {
      const procs = listProcStats()
      if (!procs) return
      const ts = Date.now()
      const wallMs = ts
      const ncpu = HOST_CPUS
      const limit = PROC_LIMIT

      /** @type {Map<string, number>} */
      const cpuByName = new Map()
      /** @type {Map<string, number>} */
      const ioByName = new Map()
      if (this._prev) {
        for (const p of procs) {
          const prev = this._prev.get(p.pid)
          if (!prev || wallMs <= prev.wallMs) continue
          const ticks = p.utime + p.stime
          const dTicks = ticks - prev.ticks
          const dSec = (wallMs - prev.wallMs) / 1000
          if (dTicks < 0 || dSec <= 0) continue
          // Linux USER_HZ typically 100
          const pct = (dTicks / 100 / dSec) * 100
          const key = p.name
          cpuByName.set(key, (cpuByName.get(key) || 0) + Math.min(100 * ncpu, pct))

          // IO is sampled every PROC_EXTRA_TICK ticks — rate over the actual sample gap
          if (
            p.ioSampleTs != null &&
            prev.ioSampleTs != null &&
            p.ioSampleTs > prev.ioSampleTs &&
            p.readBytes != null &&
            p.writeBytes != null &&
            prev.readBytes != null &&
            prev.writeBytes != null
          ) {
            const ioSec = (p.ioSampleTs - prev.ioSampleTs) / 1000
            if (ioSec > 0) {
              const dBytes =
                Math.max(0, p.readBytes - prev.readBytes) +
                Math.max(0, p.writeBytes - prev.writeBytes)
              const rateKib = dBytes / ioSec / 1024
              ioByName.set(key, (ioByName.get(key) || 0) + rateKib)
            }
          }
        }
      }

      // Hold last IO rates between sparse /proc/pid/io samples so the chart stays continuous
      if (ioByName.size) this._lastIoByName = ioByName
      const ioRates = ioByName.size ? ioByName : this._lastIoByName || ioByName

      /** @type {Map<number, { ticks: number, wallMs: number, name: string, readBytes: number|null, writeBytes: number|null, ioSampleTs: number|null }>} */
      const next = new Map()
      for (const p of procs) {
        const prev = this._prev?.get(p.pid)
        next.set(p.pid, {
          ticks: p.utime + p.stime,
          wallMs,
          name: p.name,
          readBytes: p.readBytes,
          writeBytes: p.writeBytes,
          ioSampleTs: p.ioSampleTs ?? prev?.ioSampleTs ?? null,
        })
      }
      this._prev = next

      const cpuRanked = [...cpuByName.entries()].sort((a, b) => b[1] - a[1]).slice(0, limit)
      const ioRanked = [...ioRates.entries()].sort((a, b) => b[1] - a[1]).slice(0, limit)
      const rssRanked = [...procs]
        .sort((a, b) => b.rssPages - a.rssPages)
        .slice(0, limit)
      const threadRanked = [...procs].sort((a, b) => b.threads - a.threads).slice(0, limit)

      /** @type {Map<string, number>} */
      const rssByName = new Map()
      for (const p of rssRanked) {
        const mib = (p.rssPages * this._pageBytes) / (1024 * 1024)
        rssByName.set(p.name, (rssByName.get(p.name) || 0) + mib)
      }
      const rssTop = [...rssByName.entries()].sort((a, b) => b[1] - a[1]).slice(0, limit)

      /** @type {Map<string, number>} */
      const threadsByName = new Map()
      for (const p of threadRanked) {
        threadsByName.set(p.name, (threadsByName.get(p.name) || 0) + p.threads)
      }
      const threadsTop = [...threadsByName.entries()].sort((a, b) => b[1] - a[1]).slice(0, limit)

      const cpuDef = registerTopChart(
        cpuRanked.map(([n]) => n),
        'cpu'
      )
      const rssDef = registerTopChart(
        rssTop.map(([n]) => n),
        'rss'
      )
      const ioDef = registerTopChart(
        ioRanked.map(([n]) => n),
        'io'
      )
      const threadsDef = registerTopChart(
        threadsTop.map(([n]) => n),
        'threads'
      )

      /** @type {Record<string, number|null>} */
      const cpuValues = {}
      for (const d of cpuDef.dimensions) cpuValues[d.id] = 0
      for (const [name, pct] of cpuRanked) cpuValues[name] = pct

      /** @type {Record<string, number|null>} */
      const rssValues = {}
      for (const d of rssDef.dimensions) rssValues[d.id] = 0
      for (const [name, mib] of rssTop) rssValues[name] = mib

      /** @type {Record<string, number|null>} */
      const ioValues = {}
      for (const d of ioDef.dimensions) ioValues[d.id] = 0
      for (const [name, rate] of ioRanked) ioValues[name] = rate

      /** @type {Record<string, number|null>} */
      const threadValues = {}
      for (const d of threadsDef.dimensions) threadValues[d.id] = 0
      for (const [name, count] of threadsTop) threadValues[name] = count

      this.emit('samples', [
        {
          chart: 'processes.top_cpu',
          context: 'processes.top_cpu',
          ts,
          values: cpuValues,
        },
        {
          chart: 'processes.top_rss',
          context: 'processes.top_rss',
          ts,
          values: rssValues,
        },
        {
          chart: 'processes.top_io',
          context: 'processes.top_io',
          ts,
          values: ioValues,
        },
        {
          chart: 'processes.top_threads',
          context: 'processes.top_threads',
          ts,
          values: threadValues,
        },
      ])
    } catch (err) {
      log.warn('Process tick failed', { error: err.message })
    }
    const dur = performance.now() - t0
    if (dur > 10) log.debug('tick', { collector: 'processes', durMs: Math.round(dur * 10) / 10 })
  }
}

/** @type {ProcessCollector|null} */
let singleton = null

export function getProcessCollector() {
  if (!singleton) singleton = new ProcessCollector()
  return singleton
}