Add searchable record and search 🔍

Records are added with simhashed tokens for later searching

Author: Drache93

index.js

@@ -13,6 +13,7 @@ const connect = require('./lib/connect')
 const { FIREWALL, BOOTSTRAP_NODES, KNOWN_NODES, COMMANDS } = require('./lib/constants')
 const { hash, createKeyPair } = require('./lib/crypto')
 const { decode } = require('hypercore-id-encoding')
+const { hammingDistance } = require('@holepunchto/searchable-record-cache')
 const RawStreamSet = require('./lib/raw-stream-set')
 const ConnectionPool = require('./lib/connection-pool')
 const { STREAM_NOT_CONNECTED } = require('./lib/errors')
@@ -52,6 +53,7 @@ class HyperDHT extends DHT {
     this._randomPunchInterval = opts.randomPunchInterval || 20000 // min 20s between random punches...
     this._randomPunches = 0
     this._randomPunchLimit = 1 // set to one for extra safety for now
+    this._simhash = opts.simhash
 
     this.once('persistent', () => {
       this._persistent = new Persistent(this, persistent)
@@ -184,6 +186,68 @@ class HyperDHT extends DHT {
     return this.query({ target, command: COMMANDS.LOOKUP, value: null }, opts)
   }
 
+  async searchableRecordPut(tokens, value, opts = {}) {
+    if (!this._simhash) return
+
+    const target = this._simhash.hash(tokens)
+    const query = this.query({ target, command: COMMANDS.SEARCH, value: null }, opts)
+    await query.finished()
+
+    for (const closest of query.closestReplies) {
+      await this.request(
+        {
+          target,
+          command: COMMANDS.SEARCHABLE_RECORD_PUT,
+          value: c.encode(m.searchableRecord, { value, key: target })
+        },
+        closest.from
+      )
+    }
+
+    return target
+  }
+
+  async search(tokens, opts = {}) {
+    if (!this._simhash) return
+
+    const target = this._simhash.hash(tokens)
+
+    const query = this.query(
+      {
+        target,
+        command: COMMANDS.SEARCH,
+        value: c.encode(m.searchOptions, {
+          closest: opts.closest || 5,
+          values: opts.values || 5
+        })
+      },
+      opts
+    )
+
+    const results = []
+    const seen = new Set()
+
+    for await (const reply of query) {
+      if (reply.value) {
+        const res = c.decode(m.searchResponse, reply.value)
+
+        for (const r of res) {
+          const key = r.key.toString('hex')
+          if (seen.has(key)) continue
+
+          const distance = hammingDistance(r.key, target)
+          seen.add(key)
+          results.push({ ...r, distance, from: reply.from })
+        }
+      }
+    }
+
+    results.sort((a, b) => a.distance - b.distance)
+    while (results.length > 5) results.pop()
+
+    return results
+  }
+
   lookupAndUnannounce(target, keyPair, opts = {}) {
     const unannounces = []
     const dht = this
@@ -426,6 +490,14 @@ class HyperDHT extends DHT {
         this._persistent.onimmutableget(req)
         return true
       }
+      case COMMANDS.SEARCH: {
+        this._persistent.onsearch(req)
+        return true
+      }
+      case COMMANDS.SEARCHABLE_RECORD_PUT: {
+        this._persistent.onsearchablerecordput(req)
+        return true
+      }
     }
 
     return false
@@ -604,6 +676,7 @@ function defaultCacheOpts(opts) {
         maxSize: (maxSize / 2) | 0,
         maxAge: opts.maxAge || 48 * 60 * 60 * 1000 // 48 hours
       },
+      searchableRecords: { maxSize: Infinity, maxAge: Infinity },
       bumps: { maxSize, maxAge }
     }
   }

lib/constants.js

@@ -10,7 +10,9 @@ const COMMANDS = (exports.COMMANDS = {
   MUTABLE_PUT: 6,
   MUTABLE_GET: 7,
   IMMUTABLE_PUT: 8,
-  IMMUTABLE_GET: 9
+  IMMUTABLE_GET: 9,
+  SEARCHABLE_RECORD_PUT: 10,
+  SEARCH: 11
 })
 
 exports.BOOTSTRAP_NODES = global.Pear?.config.dht?.bootstrap || [

lib/messages.js

@@ -420,3 +420,56 @@ exports.mutableGetResponse = {
     }
   }
 }
+
+exports.searchOptions = {
+  preencode(state, m) {
+    c.uint.preencode(state, m.closest)
+    c.uint.preencode(state, m.values)
+  },
+  encode(state, m) {
+    c.uint.encode(state, m.closest)
+    c.uint.encode(state, m.values)
+  },
+  decode(state) {
+    return {
+      closest: c.uint.decode(state),
+      values: c.uint.decode(state)
+    }
+  }
+}
+
+exports.searchableRecord = {
+  preencode(state, m) {
+    c.fixed32.preencode(state, m.key)
+    c.fixed32.preencode(state, m.value)
+  },
+  encode(state, m) {
+    c.fixed32.encode(state, m.key)
+    c.fixed32.encode(state, m.value)
+  },
+  decode(state) {
+    return {
+      key: c.fixed32.decode(state),
+      value: c.fixed32.decode(state)
+    }
+  }
+}
+
+const searchResponseItem = {
+  preencode(state, m) {
+    c.fixed32.preencode(state, m.key)
+    c.array(c.fixed32).preencode(state, m.values)
+  },
+  encode(state, m) {
+    c.fixed32.encode(state, m.key)
+    c.array(c.fixed32).encode(state, m.values)
+  },
+  decode(state) {
+    return {
+      key: c.fixed32.decode(state),
+      values: c.array(c.fixed32).decode(state)
+    }
+  }
+}
+
+exports.searchResponse = c.array(searchResponseItem)

lib/persistent.js

@@ -1,6 +1,7 @@
 const c = require('compact-encoding')
 const sodium = require('sodium-universal')
 const RecordCache = require('record-cache')
+const { SearchableRecordCache } = require('@holepunchto/searchable-record-cache')
 const Cache = require('xache')
 const b4a = require('b4a')
 const unslab = require('unslab')
@@ -21,6 +22,7 @@ module.exports = class Persistent {
     this.refreshes = new Cache(opts.refreshes)
     this.mutables = new Cache(opts.mutables)
     this.immutables = new Cache(opts.immutables)
+    this.searchableRecords = new SearchableRecordCache(opts.searchableRecords)
   }
 
   onlookup(req) {
@@ -226,11 +228,47 @@ module.exports = class Persistent {
     req.reply(null)
   }
 
+  async onsearchablerecordput(req) {
+    if (!req.target) return
+
+    const doc = c.decode(m.searchableRecord, req.value)
+    const { key, value } = doc
+
+    this.searchableRecords.add(key, value)
+
+    req.reply(null)
+  }
+
+  onsearch(req) {
+    if (!req.target || !req.value) {
+      // allow searchable record put to do an empty search
+      req.reply(null)
+      return
+    }
+
+    const opts = c.decode(m.searchOptions, req.value)
+    const results = []
+
+    const res = this.searchableRecords.search(req.target, opts)
+
+    for (const r of res) {
+      results.push({
+        values: r.values,
+        key: r.key,
+        distance: r.distance
+      })
+    }
+
+    results.sort((a, b) => a.distance - b.distance)
+    req.reply(c.encode(m.searchResponse, results.slice(0, 12))) // max we can fit
+  }
+
   destroy() {
     this.records.destroy()
     this.refreshes.destroy()
     this.mutables.destroy()
     this.immutables.destroy()
+    this.searchableRecords.destroy()
   }
 
   static signMutable(seq, value, keyPair) {

package.json

@@ -25,6 +25,8 @@
     }
   },
   "dependencies": {
+    "@holepunchto/searchable-record-cache": "^0.0.6",
+    "@holepunchto/simhash": "^0.0.6",
     "@hyperswarm/secret-stream": "^6.6.2",
     "b4a": "^1.3.1",
     "bare-events": "^2.2.0",

test/all.js

@@ -16,6 +16,7 @@ async function runTests() {
   await import('./noncustodial.js')
   await import('./pool.js')
   await import('./relaying.js')
+  await import('./search.js')
   await import('./storing.js')
 
   test.resume()

test/messages.js

@@ -440,3 +440,26 @@ test('announce with refresh', function (t) {
 
   t.alike(d, ann)
 })
+
+test('search', function (t) {
+  const state = { start: 0, end: 0, buffer: null }
+
+  const searchResponse = []
+  for (let i = 0; i < 12; i++) {
+    searchResponse.push({
+      key: Buffer.alloc(32).fill(i),
+      values: [Buffer.alloc(32).fill(i)]
+    })
+  }
+
+  m.searchResponse.preencode(state, searchResponse)
+  state.buffer = b4a.allocUnsafe(state.end)
+  m.searchResponse.encode(state, searchResponse)
+
+  t.is(state.end, state.start, 'fully encoded')
+
+  state.start = 0
+  const d = m.searchResponse.decode(state)
+
+  t.alike(d, searchResponse)
+})