modules/measurement-rate-control/hyper-p2p-histogram-gossip/index.js

require('bare-process/global')
const EventEmitter = require('bare-events')
const b4a = require('b4a')
const { assertNonEmpty } = require('../../_shared/lib/errors.js')
const { initModuleSwarm, gossipSend } = require('../../_shared/p2p-bare.js')

const PROTOCOL = 'histogram-gossip/v1'
const DEFAULT_BUCKETS = 10

function bucketIndex (value, min, max, count) {
  if (value <= min) return 0
  if (value >= max) return count - 1
  const span = (max - min) / count
  return Math.min(count - 1, Math.floor((value - min) / span))
}

class HyperP2PHistogramGossip extends EventEmitter {
  constructor (opts = {}) {
    super()
    this.topic = opts.topic || null
    this.keyPair = opts.keyPair || require('hypercore-crypto').keyPair()
    this.peerHex = b4a.toString(this.keyPair.publicKey, 'hex')
    this._bucketCount = opts.buckets || DEFAULT_BUCKETS
    this._metrics = new Map()
    this._stats = { observed: 0, gossipIn: 0, gossipOut: 0 }
    this.swarm = null
    this._peerMsgs = null
  }

  _ensure (metric) {
    if (!this._metrics.has(metric)) {
      this._metrics.set(metric, {
        min: Infinity,
        max: -Infinity,
        counts: new Array(this._bucketCount).fill(0),
        total: 0
      })
    }
    return this._metrics.get(metric)
  }

  observe (metric, value) {
    assertNonEmpty(metric, 'metric')
    if (typeof value !== 'number' || Number.isNaN(value)) {
      throw new Error('value must be a number')
    }
    const h = this._ensure(metric)
    h.total++
    if (value < h.min) h.min = value
    if (value > h.max) h.max = value
    if (h.min === h.max) {
      h.counts[0]++
    } else {
      const idx = bucketIndex(value, h.min, h.max, this._bucketCount)
      h.counts[idx]++
    }
    this._stats.observed++
    this._gossip({
      type: 'histogram-buckets',
      metric,
      min: h.min,
      max: h.max,
      counts: [...h.counts],
      peer: this.peerHex,
      at: Date.now()
    })
    this.emit('observe', { metric, value })
    return true
  }

  percentile (metric, p) {
    assertNonEmpty(metric, 'metric')
    const h = this._metrics.get(metric)
    if (!h || !h.total) return null
    const pct = Math.max(0, Math.min(100, Number(p)))
    const target = Math.ceil((pct / 100) * h.total)
    let seen = 0
    for (let i = 0; i < h.counts.length; i++) {
      seen += h.counts[i]
      if (seen >= target) {
        if (h.min === h.max) return h.min
        const span = (h.max - h.min) / h.counts.length
        return h.min + span * (i + 0.5)
      }
    }
    return h.max
  }

  snapshot (metric) {
    const h = this._metrics.get(metric)
    if (!h) return null
    return { metric, min: h.min, max: h.max, counts: [...h.counts], total: h.total }
  }

  metrics () {
    return [...this._metrics.keys()].sort()
  }

  resetMetric (metric) {
    return this._metrics.delete(String(metric))
  }

  clearAll () {
    const n = this._metrics.size
    this._metrics.clear()
    return n
  }

  observeBatch (entries) {
    if (!Array.isArray(entries)) throw new Error('entries array required')
    return entries.map((e) => this.observe(e.metric, e.value))
  }

  totalObservations () {
    let n = 0
    for (const h of this._metrics.values()) n += h.total
    return n
  }

  _mergeBuckets (local, remote) {
    if (!remote || !remote.counts) return
    if (remote.min < local.min) local.min = remote.min
    if (remote.max > local.max) local.max = remote.max
    const len = Math.min(local.counts.length, remote.counts.length)
    for (let i = 0; i < len; i++) local.counts[i] += remote.counts[i]
    local.total += remote.counts.reduce((a, b) => a + b, 0)
  }

  _gossip (data) {
    if (!this._peerMsgs) return
    gossipSend(this, data)
    this._stats.gossipOut++
  }

  _onGossip (data) {
    if (!data || data.type !== 'histogram-buckets' || !data.metric) return
    this._stats.gossipIn++
    const h = this._ensure(data.metric)
    this._mergeBuckets(h, data)
    this.emit('remote-buckets', { metric: data.metric, peer: data.peer })
  }

  getStats () {
    return {
      ...this._stats,
      metrics: this._metrics.size,
      protocol: PROTOCOL
    }
  }

  async ready () {
    if (this.swarm || !this.topic) return this
    await initModuleSwarm(this, {
      keyPair: this.keyPair,
      topic: this.topic,
      protocol: PROTOCOL,
      onmessage: (data) => this._onGossip(data)
    })
    return this
  }

  async close () {
    this._metrics.clear()
    if (this.swarm) await this.swarm.destroy().catch(() => {})
    this.swarm = null
    this.emit('closed')
  }
}

module.exports = { HyperP2PHistogramGossip, PROTOCOL }