modules/indexes-search/hyper-p2p-bloom-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 = 'bloom-gossip/v1'
const DEFAULT_BITS = 2048
const HASH_COUNT = 4

function fnvHash (str, seed) {
  let h = seed >>> 0
  for (let i = 0; i < str.length; i++) {
    h ^= str.charCodeAt(i)
    h = Math.imul(h, 16777619)
  }
  return h >>> 0
}

function positions (key, size, k) {
  const out = []
  for (let i = 0; i < k; i++) out.push(fnvHash(key, 0x811c9dc5 + i * 31) % size)
  return out
}

class HyperP2PBloomGossip 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._size = opts.bits || DEFAULT_BITS
    this._bits = new Uint8Array(Math.ceil(this._size / 8))
    this._keys = new Set()
    this._stats = { added: 0, queries: 0, gossipIn: 0, gossipOut: 0 }
    this.swarm = null
    this._peerMsgs = null
  }

  add (key) {
    assertNonEmpty(key, 'key')
    const k = String(key)
    if (this._keys.has(k)) return false
    for (const pos of positions(k, this._size, HASH_COUNT)) {
      const byte = pos >> 3
      const bit = pos & 7
      this._bits[byte] |= 1 << bit
    }
    this._keys.add(k)
    this._stats.added++
    this._gossip({ type: 'bloom-add', key: k, peer: this.peerHex, at: Date.now() })
    this.emit('add', { key: k })
    return true
  }

  mightContain (key) {
    assertNonEmpty(key, 'key')
    this._stats.queries++
    const k = String(key)
    for (const pos of positions(k, this._size, HASH_COUNT)) {
      const byte = pos >> 3
      const bit = pos & 7
      if (!(this._bits[byte] & (1 << bit))) return false
    }
    return true
  }

  exportBits () {
    return {
      size: this._size,
      hashCount: HASH_COUNT,
      bits: b4a.toString(this._bits, 'hex'),
      count: this._keys.size
    }
  }

  importBits (payload) {
    if (!payload || !payload.bits) return false
    this._bits = b4a.from(payload.bits, 'hex')
    if (payload.size) this._size = payload.size
    return true
  }

  size () {
    return this._keys.size
  }

  listKeys () {
    return [...this._keys].sort()
  }

  clear () {
    this._bits.fill(0)
    const n = this._keys.size
    this._keys.clear()
    this._stats.added = 0
    this.emit('clear', { count: n })
    return n
  }

  union (payload) {
    if (!payload || !payload.bits) return 0
    const remote = b4a.from(payload.bits, 'hex')
    let merged = 0
    for (let i = 0; i < Math.min(this._bits.length, remote.length); i++) {
      const before = this._bits[i]
      this._bits[i] |= remote[i]
      if (this._bits[i] !== before) merged++
    }
    if (payload.size) this._size = payload.size
    this.emit('union', { mergedBytes: merged })
    return merged
  }

  fillRatio () {
    let set = 0
    for (let i = 0; i < this._bits.length; i++) {
      for (let bit = 0; bit < 8; bit++) {
        if (this._bits[i] & (1 << bit)) set++
      }
    }
    return this._size ? set / this._size : 0
  }

  addBatch (keys) {
    if (!Array.isArray(keys)) throw new Error('keys array required')
    let n = 0
    for (const k of keys) if (this.add(k)) n++
    return n
  }

  filterMaybeNew (keys) {
    if (!Array.isArray(keys)) throw new Error('keys array required')
    return keys.filter((k) => !this.mightContain(k))
  }

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

  _onGossip (data) {
    if (!data || data.type !== 'bloom-add' || !data.key) return
    this._stats.gossipIn++
    if (!this._keys.has(data.key)) {
      for (const pos of positions(data.key, this._size, HASH_COUNT)) {
        const byte = pos >> 3
        const bit = pos & 7
        this._bits[byte] |= 1 << bit
      }
      this._keys.add(data.key)
      this.emit('remote-add', { key: data.key, peer: data.peer })
    }
  }

  getStats () {
    return {
      ...this._stats,
      bits: this._size,
      keys: this._keys.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._bits.fill(0)
    this._keys.clear()
    if (this.swarm) await this.swarm.destroy().catch(() => {})
    this.swarm = null
    this.emit('closed')
  }
}

module.exports = { HyperP2PBloomGossip, PROTOCOL }