Add decrypted api for for box2 decrypted messages

README.md

@@ -473,6 +473,40 @@ operators:
 See [jitdb operators] and [operators/index.js] for a complete list of supported
 operators.
 
+### decrypted
+
+A pull-stream source with newly decrypted values returned as full
+messages. JITDB doesn't have a concept of exactly what values are
+decrypted so a separate api is needed. This api can be combined with
+`where` to receive old values, live values or decrypted values. 
+
+Example:
+
+``` js
+    const pull = require('pull-stream')
+    const cat = require('pull-cat')
+
+    pull(
+      cat([
+        sbot2.db.query(
+          where(type('post')),
+          toPullStream()
+        ),
+        pull(
+          sbot2.db.decrypted,
+          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)

core.js

@@ -978,6 +978,22 @@ exports.init = function (sbot, config) {
 
   const reindexingLock = mutexify()
 
+  function LazyPull() {
+    let onValue
+
+    return {
+      add(val) {
+        if (onValue) onValue(null, val)
+      },
+      source(abort, cb) {
+        if (abort) cb(abort)
+        else if (!onValue) onValue = cb
+      },
+    }
+  }
+
+  const decryptedValues = LazyPull()
+
   function reindexEncrypted(cb) {
     indexingActive.set(indexingActive.value + 1)
     reindexingLock((unlock) => {
@@ -1008,6 +1024,8 @@ exports.init = function (sbot, config) {
 
             const pValue = bipf.seekKey2(buf, 0, BIPF_VALUE, 0)
 
+            decryptedValues.add(bipf.decode(buf, 0))
+
             onDrain('keys', () => {
               indexes['keys'].getSeq(key, (err, seq) => {
                 // prettier-ignore
@@ -1120,6 +1138,7 @@ exports.init = function (sbot, config) {
     getMsg,
     query,
     prepare,
+    decrypted: decryptedValues.source,
     del,
     deleteFeed,
     add,

test/query-decrypted-box2.js

@@ -0,0 +1,109 @@
+// 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 cat = require('pull-cat')
+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(
+      cat([
+        sbot2.db.query(
+          where(type('post')),
+          toPullStream()
+        ),
+        pull(
+          sbot2.db.decrypted,
+          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')
+            })
+          })
+        )
+      )
+    })
+  })
+})