/**
 * Kafka JMX/exporter HTTP probe (Burrow-style or prometheus jmx exporter).
 * Enable: PEARDATA_KAFKA=1
 * URL: PEARDATA_KAFKA_URL=http://127.0.0.1:9308/metrics (prometheus) OR
 *      PEARDATA_KAFKA_URL=http://127.0.0.1:8080 (plain JSON under /stats)
 */
import { CollectorPlugin } from './plugin.js'
import { registerChart } from '../../../shared/metrics.js'

export function isKafkaEnabled() {
  const v = process.env.PEARDATA_KAFKA
  return v === '1' || v === 'on' || v === 'true'
}

function url() {
  return (process.env.PEARDATA_KAFKA_URL || 'http://127.0.0.1:9308/metrics').replace(/\/$/, '')
}

export class KafkaCollector extends CollectorPlugin {
  constructor() {
    super({ name: 'kafka' })
  }

  isEnabled() {
    return isKafkaEnabled()
  }

  async collect() {
    const ts = Date.now()
    registerChart({
      id: 'kafka.up',
      name: 'kafka.up',
      context: 'kafka.up',
      title: 'Kafka exporter up',
      units: 'boolean',
      family: 'kafka',
      chartType: 'line',
      priority: 8800,
      plugin: 'kafka',
      dimensions: [{ id: 'up', name: 'up', algorithm: 'absolute' }],
    })
    try {
      const res = await fetch(url(), { signal: AbortSignal.timeout(2500) })
      if (!res.ok) throw new Error(String(res.status))
      const text = await res.text()
      /** @type {Record<string, number>} */
      const vals = { brokers: 0, under_replicated: 0, offline_count: 0 }
      for (const line of text.split('\n')) {
        if (line.startsWith('#') || !line.trim()) continue
        const m = line.match(/^([a-zA-Z0-9_:]+)(?:\{[^}]*\})?\s+([0-9.eE+-]+)/)
        if (!m) continue
        const name = m[1]
        const v = Number(m[2])
        if (name.includes('broker') && name.includes('info')) vals.brokers += 1
        if (name.includes('under_replicated')) vals.under_replicated += v
        if (name.includes('topic') && name.includes('partitions')) vals.partition_count += v
      }
      registerChart({
        id: 'kafka.cluster',
        name: 'kafka.cluster',
        context: 'kafka.cluster',
        title: 'Kafka cluster',
        units: 'count',
        family: 'kafka',
        chartType: 'line',
        priority: 8810,
        plugin: 'kafka',
        dimensions: Object.keys(vals).map((id) => ({
          id,
          name: id,
          algorithm: 'absolute',
        })),
      })
      return [
        { chart: 'kafka.up', context: 'kafka.up', ts, values: { up: 1 } },
        { chart: 'kafka.cluster', context: 'kafka.cluster', ts, values: vals },
      ]
    } catch {
      return [{ chart: 'kafka.up', context: 'kafka.up', ts, values: { up: 0 } }]
    }
  }
}

let singleton = null
export function getKafkaCollector() {
  if (!singleton) singleton = new KafkaCollector()
  return singleton
}