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() } _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 }