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 = 'inverted-index/v1' function normalizeTerm (term) { return String(term).toLowerCase().trim() } class HyperP2PInvertedIndex 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._terms = new Map() this._docs = new Map() this._stats = { indexed: 0, removed: 0, queries: 0, gossipIn: 0, gossipOut: 0 } this.swarm = null this._peerMsgs = null } index (docId, terms) { assertNonEmpty(docId, 'docId') const normalized = (terms || []).map(normalizeTerm).filter(Boolean) const prev = this._docs.get(docId) || new Set() for (const term of normalized) { const set = this._terms.get(term) || new Set() set.add(docId) this._terms.set(term, set) if (!prev.has(term)) { this._gossip({ type: 'term-index-sync', op: 'add', term, docId, peer: this.peerHex, at: Date.now() }) } } this._docs.set(docId, new Set(normalized)) this._stats.indexed++ this.emit('index', { docId, terms: normalized }) return { docId, terms: normalized } } search (term) { this._stats.queries++ const set = this._terms.get(normalizeTerm(term)) return set ? [...set].sort() : [] } searchAll (terms, mode = 'and') { if (!Array.isArray(terms) || !terms.length) return [] const normalized = terms.map(normalizeTerm).filter(Boolean) if (!normalized.length) return [] this._stats.queries++ if (mode === 'or') { const out = new Set() for (const term of normalized) { for (const docId of this.search(term)) out.add(docId) } return [...out].sort() } let hits = null for (const term of normalized) { const set = this._terms.get(term) const ids = set ? [...set] : [] if (hits === null) hits = new Set(ids) else { const next = new Set() for (const id of ids) if (hits.has(id)) next.add(id) hits = next } } return hits ? [...hits].sort() : [] } getDocTerms (docId) { const terms = this._docs.get(docId) return terms ? [...terms].sort() : [] } removeDoc (docId) { assertNonEmpty(docId, 'docId') const terms = this._docs.get(docId) if (!terms) return false for (const term of terms) { const set = this._terms.get(term) if (set) { set.delete(docId) if (!set.size) this._terms.delete(term) } this._gossip({ type: 'term-index-sync', op: 'remove', term, docId, peer: this.peerHex, at: Date.now() }) } this._docs.delete(docId) this._stats.removed++ this.emit('remove', { docId }) return true } listTerms () { return [...this._terms.keys()].sort() } docCount () { return this._docs.size } hasDoc (docId) { return this._docs.has(String(docId)) } topTerms (limit = 10) { const ranked = [...this._terms.entries()] .map(([term, set]) => ({ term, docs: set.size })) .sort((a, b) => b.docs - a.docs) return ranked.slice(0, Math.max(0, limit | 0)) } _gossip (data) { if (!this._peerMsgs) return gossipSend(this, data) this._stats.gossipOut++ } _onGossip (data) { if (!data || data.type !== 'term-index-sync') return this._stats.gossipIn++ const term = normalizeTerm(data.term) const docId = data.docId if (!term || !docId) return if (data.op === 'remove') { const set = this._terms.get(term) if (set) { set.delete(docId) if (!set.size) this._terms.delete(term) } const docTerms = this._docs.get(docId) if (docTerms) docTerms.delete(term) this.emit('remote-remove', { term, docId, peer: data.peer }) return } const set = this._terms.get(term) || new Set() set.add(docId) this._terms.set(term, set) const docTerms = this._docs.get(docId) || new Set() docTerms.add(term) this._docs.set(docId, docTerms) this.emit('remote-index', { term, docId, peer: data.peer }) } getStats () { return { ...this._stats, terms: this._terms.size, docs: this._docs.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._terms.clear() this._docs.clear() if (this.swarm) await this.swarm.destroy().catch(() => {}) this.swarm = null this.emit('closed') } } module.exports = { HyperP2PInvertedIndex, PROTOCOL }