fix branchStream to restart after box2 decryption

lookup.js

@@ -53,6 +53,7 @@ exports.init = function (sbot, config) {
   let stateLoaded = false
   let loadStateRequested = false
   let liveDrainer = null
+  let removeIndexingActiveListener = null
   let notifyNewBranch = null
   let mySeed = null
   const detailsLookup = new Map() // feedId => FeedDetails
@@ -170,19 +171,30 @@ exports.init = function (sbot, config) {
 
             sbot.close.hook(function (fn, args) {
               if (liveDrainer) liveDrainer.abort(true)
+              if (removeIndexingActiveListener) removeIndexingActiveListener()
               if (notifyNewBranch) notifyNewBranch.abort(true)
               fn.apply(this, args)
             })
 
-            pull(
-              sbot.db.query(
-                where(authorIsBendyButtV1()),
-                live(),
-                toPullStream()
-              ),
-              pull.filter((msg) => validate.isValid(msg)),
-              (liveDrainer = pull.drain(updateLookupFromMsg))
-            )
+            // We need to wait for the indexing (including box2 reindexing
+            // encrypted messages) to be done before we can read the messages
+            const obz = sbot.db.getIndexingActive()
+            removeIndexingActiveListener = obz((indexingActive) => {
+              if (indexingActive !== 0) {
+                if (liveDrainer) liveDrainer.abort(true)
+                return
+              }
+
+              pull(
+                sbot.db.query(
+                  where(authorIsBendyButtV1()),
+                  live({ old: true }),
+                  toPullStream()
+                ),
+                pull.filter((msg) => validate.isValid(msg)),
+                (liveDrainer = pull.drain(updateLookupFromMsg))
+              )
+            })
           })
         )
       })

test/api/branch-stream.test.js

@@ -4,8 +4,11 @@
 
 const test = require('tape')
 const pull = require('pull-stream')
+const ssbKeys = require('ssb-keys')
+const p = require('util').promisify
 const Testbot = require('../testbot')
 const { setupTree, testReadAndPersisted } = require('../testtools')
+const keysAPI = require('../../keys')
 
 test('branchStream', (t) => {
   const sbot = Testbot()
@@ -183,3 +186,106 @@ test('branchStream (encrypted announces)', (t) => {
     )
   })
 })
+
+test('branchStream can reprocess encrypted announces', async (t) => {
+  const mainKeys = ssbKeys.generate(null, 'alice')
+  const metafeedSeed = Buffer.from(
+    '000000000000000000000000000000000000000000000000000000000000beef',
+    'hex'
+  )
+  const metafeedKeys = keysAPI.deriveRootMetaFeedKeyFromSeed(metafeedSeed)
+
+  const sbot = Testbot({ keys: mainKeys, metafeedSeed })
+
+  const ownKey1 = Buffer.from(
+    '30720d8f9cbf37f6d7062826f6decac93e308060a8aaaa77e6a4747f40ee1a76',
+    'hex'
+  )
+
+  const ownKey2 = Buffer.from(
+    'abcd0d8f9cbf37f6d7062826f6decac93e308060a8aaaa77e6a4747f40ee1a76',
+    'hex'
+  )
+
+  // Guarantee we have a root metafeed and the main feed linked to it
+  const root = await p(sbot.metafeeds.findOrCreate)()
+  // Then pluck the shard feed so we can manually add to it
+  const any = () => true
+  const v1 = await p(sbot.metafeeds.advanced.findOrCreate)(root, any, null)
+  const shard = await p(sbot.metafeeds.advanced.findOrCreate)(v1, any, null)
+  t.pass('set up root/v1/:shard/main')
+
+  sbot.box2.setOwnDMKey(ownKey1)
+  sbot.box2.addKeypair(root.keys)
+  sbot.box2.addKeypair(v1.keys)
+  sbot.box2.addKeypair(shard.keys)
+  sbot.box2.addKeypair(mainKeys)
+
+  const opts = sbot.metafeeds.messages.optsForAddDerived(
+    shard.keys,
+    'group',
+    metafeedSeed,
+    'classic',
+    {},
+    [metafeedKeys.id],
+    'box2'
+  )
+  await p(sbot.db.create)(opts)
+  t.pass('published encrypted announce message on a shard')
+
+  sbot.box2.setOwnDMKey(ownKey2)
+  t.pass('change own DM key so that branchStream cannot decrypt')
+
+  let expectedLive = ['root/v1/2/group']
+  pull(
+    sbot.metafeeds.branchStream({old: false, live: true}),
+    pull.drain((branch) => {
+      const path = branch.map((f) => f.purpose).join('/')
+      t.equals(path, expectedLive.shift(), 'branchStream can decrypt announce')
+    })
+  )
+
+  await new Promise((resolve) => {
+    pull(
+      sbot.metafeeds.branchStream({ old: true, live: false }),
+      pull.collect((err, branches) => {
+        if (err) t.error(err, 'no error')
+        const summary = branches.map((b) => b.map((f) => f.purpose).join('/'))
+        t.deepEquals(
+          summary,
+          ['root', 'root/v1', 'root/v1/2', 'root/v1/2/main'],
+          'branchStream shows that group subfeed is missing'
+        )
+        resolve()
+      })
+    )
+  })
+
+  sbot.box2.setOwnDMKey(ownKey1)
+  t.pass('changed own DM key back so that branchStream CAN decrypt')
+
+  await p(sbot.db.reindexEncrypted)()
+  t.pass('reindexed encrypted messages')
+
+  await new Promise((resolve) => {
+    pull(
+      sbot.metafeeds.branchStream({ old: true, live: false }),
+      pull.collect((err, branches) => {
+        if (err) t.error(err, 'no error')
+        const summary = branches.map((b) => b.map((f) => f.purpose).join('/'))
+        t.deepEquals(
+          summary,
+          ['root', 'root/v1', 'root/v1/2', 'root/v1/2/main', 'root/v1/2/group'],
+          'branchStream shows that group subfeed is present'
+        )
+        resolve()
+      })
+    )
+  })
+
+  while (expectedLive.length > 0) {
+    await p(setTimeout)(100)
+  }
+
+  await p(sbot.close)(true)
+})

test/testbot.js

@@ -10,11 +10,13 @@ const caps = require('ssb-caps')
 
 let count = 0
 
-// opts.path      (optional)
-//   opts.name    (optional) - convenience method for deterministic opts.path
-// opts.keys      (optional)
-// opts.rimraf    (optional) - clear the directory before start (default: true)
-
+/**
+ * - opts.path (optional)
+ *   - opts.name (optional) - convenience method for deterministic opts.path
+ * - opts.rimraf (optional) - clear the directory before start (default: true)
+ * - opts.keys (optional)
+ * - opts.metafeedSeed (optional)
+ */
 module.exports = function createSbot(opts = {}) {
   const dir = opts.path || `/tmp/metafeeds-metafeed-${opts.name || count++}`
   if (opts.rimraf !== false) rimraf.sync(dir)
@@ -32,10 +34,12 @@ module.exports = function createSbot(opts = {}) {
     path: dir,
     keys,
     metafeeds: {
-      seed: Buffer.from(
-        '000000000000000000000000000000000000000000000000000000000000beef',
-        'hex'
-      ),
+      seed:
+        opts.metafeedSeed ||
+        Buffer.from(
+          '000000000000000000000000000000000000000000000000000000000000beef',
+          'hex'
+        ),
     },
   })
 }