Add reindexed api for for box2 decrypted messages (#410)

* Add test for box2 decryption and live queries

* Add reindexed pull source api

README.md

@@ -473,6 +473,44 @@ operators:
 See [jitdb operators] and [operators/index.js] for a complete list of supported
 operators.
 
+### reindexed()
+
+A pull-stream source of newly decrypted reindexed values returned as
+full messages. Calling `reindexEncrypted` after adding a new box2 key
+will trigger new values in this stream.
+
+This api can be combined with `where` to receive messages of a
+particular type no matter if they are existing, newly added or
+decrypted values.
+
+Example of getting existing post messages together with newly
+decrypted ones:
+
+``` js
+    const pull = require('pull-stream')
+    const cat = require('pull-cat')
+
+    pull(
+      cat([
+        sbot2.db.query(
+          where(type('post')),
+          toPullStream()
+        ),
+        pull(
+          sbot2.db.reindexed(),
+          pull.filter((msg) => {
+            return msg.value.content.type === 'post'
+          })
+        )
+      ]),
+      pull.drain(
+        (result) => {
+           console.log("got a new post", result.value)
+        }
+      )
+    )
+```
+
 ### add(nativeMsg, cb)
 
 Validate and add a message to the database. The callback will the (possible)
@@ -564,6 +602,8 @@ at runtime and that changes what messages can be decrypted. Calling
 this function is needed after adding a new key. The function can be
 called multiple times safely.
 
+The decrypted values from this can be consumed using `reindexed`.
+
 ### logStats(cb)
 
 Use [async-append-only-log]'s `stats` method to get information on how many

core.js

@@ -977,6 +977,7 @@ exports.init = function (sbot, config) {
   }
 
   const reindexingLock = mutexify()
+  const reindexedValues = Notify()
 
   function reindexEncrypted(cb) {
     indexingActive.set(indexingActive.value + 1)
@@ -1008,6 +1009,8 @@ exports.init = function (sbot, config) {
 
             const pValue = bipf.seekKey2(buf, 0, BIPF_VALUE, 0)
 
+            reindexedValues(bipf.decode(buf, 0))
+
             onDrain('keys', () => {
               indexes['keys'].getSeq(key, (err, seq) => {
                 // prettier-ignore
@@ -1133,6 +1136,7 @@ exports.init = function (sbot, config) {
     onMsgAdded,
     compact,
     reindexEncrypted,
+    reindexed: () => reindexedValues.listen(),
     logStats: log.stats,
     indexingProgress: () => indexingProgress.listen(),
     compactionProgress: () => compactionProgress.listen(),

test/query-decrypted-box2.js

@@ -0,0 +1,100 @@
+// SPDX-FileCopyrightText: 2022 Anders Rune Jensen
+//
+// SPDX-License-Identifier: Unlicense
+
+const test = require('tape')
+const ssbKeys = require('ssb-keys')
+const path = require('path')
+const rimraf = require('rimraf')
+const mkdirp = require('mkdirp')
+const SecretStack = require('secret-stack')
+const caps = require('ssb-caps')
+const ssbUri = require('ssb-uri2')
+const pull = require('pull-stream')
+const fs = require('fs')
+const {
+  where,
+  type,
+  toPullStream,
+  toCallback
+} = require('../operators')
+
+const dir = '/tmp/ssb-db2-query-decrypted-handle-box2'
+rimraf.sync(dir)
+mkdirp.sync(dir)
+
+const dir2 = '/tmp/ssb-db2-query-decrypted-handle-box2-2'
+rimraf.sync(dir2)
+mkdirp.sync(dir2)
+
+test('decrypted api contains newly decrypted box2 messages', (t) => {
+  const keys = ssbKeys.loadOrCreateSync(path.join(dir, 'secret'))
+  const sbot = SecretStack({ appKey: caps.shs })
+    .use(require('../'))
+    .call(null, { keys, path: dir })
+
+  const testkey = Buffer.from(
+    '30720d8f9cbf37f6d7062826f6decac93e308060a8aaaa77e6a4747f40ee1a76',
+    'hex'
+  )
+  const groupId = ssbUri.compose({
+    type: 'identity',
+    format: 'group',
+    data: '-oaWWDs8g73EZFUMfW37R_ULtFEjwKN_DczvdYihjbU=',
+  })
+  sbot.box2.addGroupInfo(groupId, { key: testkey })
+
+  const post = {
+    feedFormat: 'classic',
+    content: { type: 'post', text: 'Testing!' },
+    recps: [groupId],
+    encryptionFormat: 'box2',
+  }
+
+  sbot.db.create(post, (err, msgBoxed) => {
+    t.error(err, 'no err')
+    t.equal(typeof msgBoxed.value.content, 'string')
+    t.true(msgBoxed.value.content.endsWith('.box2'), '.box2')
+
+    const keys2 = ssbKeys.loadOrCreateSync(path.join(dir2, 'secret'))
+    const sbot2 = SecretStack({ appKey: caps.shs })
+      .use(require('../'))
+      .call(null, { keys: keys2, path: dir2 })
+
+    // setup decrypted handler
+    pull(
+      sbot2.db.reindexed(),
+      pull.filter((msg) => {
+        return msg.value.content.type === 'post'
+      }),
+      pull.drain(
+        (result) => {
+          t.equal(result.value.content.text, 'Testing!')
+          sbot.close(() => {
+            sbot2.close(t.end)
+          })
+        }
+      )
+    )
+
+    sbot2.db.add(msgBoxed.value, (err) => {
+      t.error(err, 'no err')
+
+      // make sure we have queries indexed before adding the key
+      pull(
+        sbot2.db.query(
+          where(type('post')),
+          toCallback((err, results) => {
+            t.error(err, 'no err')
+            t.equal(results.length, 0, 'no results')
+
+            sbot2.box2.addGroupInfo(groupId, { key: testkey })
+            sbot2.db.reindexEncrypted((err) => {
+              t.error(err, 'no err')
+            })
+          })
+        )
+      )
+    })
+  })
+})