merge master

Author: mixmix

lib/epochs/index.js

@@ -3,17 +3,26 @@
 // SPDX-License-Identifier: LGPL-3.0-only
 
 const { promisify: p } = require('util')
-const { fromMessageSigil } = require('ssb-uri2')
+const { fromMessageSigil, toMessageSigil } = require('ssb-uri2')
 const pull = require('pull-stream')
 const pullDefer = require('pull-defer')
+const pullMany = require('pull-many')
 const pullFlatMerge = require('pull-flat-merge')
 const Strategy = require('@tangle/strategy')
 const Reduce = require('@tangle/reduce')
 const OverwriteFields = require('@tangle/overwrite-fields')
 const clarify = require('clarify-error')
 const Butt64 = require('butt64')
 const isCanonicalBase64 = require('is-canonical-base64')
-const { where, and, type, live, toPullStream } = require('ssb-db2/operators')
+const {
+  where,
+  and,
+  type,
+  live,
+  key,
+  isDecrypted,
+  toPullStream,
+} = require('ssb-db2/operators')
 const {
   validator: {
     group: {
@@ -29,9 +38,9 @@ const isSubsetOf = require('set.prototype.issubsetof')
 const intersection = require('set.prototype.intersection')
 
 const GetMembers = require('./get-members')
-
-const { groupRecp } = require('../operators')
 const hookClose = require('../hook-close')
+const { groupRecp } = require('../operators')
+
 const getTangleUpdates = require('../tangles/get-tangle-updates')
 
 const msgPattern = toPattern(new Butt64('ssb:message/[a-zA-Z0-9-]+/', null, 32))
@@ -179,14 +188,16 @@ module.exports = function Epochs(ssb) {
     // then skip all the preferrentEpochs until we get up to the current
     // This is important for listMembers to not send confusing results
     getPreferredEpoch(groupId, (err, preferredEpoch) => {
-      if (err) return deferredSource.abort(clarify(err, 'failed to get initial preferred epoch'))
-
+      // if we're live and this fails we don't really mind, just go straight to live
       if (!live) {
-        deferredSource.resolve(pull.once(preferredEpoch))
+        if (err) deferredSource.abort(clarify(err, 'failed to get initial preferred epoch'))
+        else deferredSource.resolve(pull.once(preferredEpoch))
+
         return
       }
 
-      var sync = false
+      // if we couldn't get current preferred, we'll just go live
+      var sync = !!err
       const source = pull(
         epochsReduce.stream(ssb, groupId, { getters: allGetters, live }),
         pull.asyncMap(BuildPreferredEpoch(ssb, groupId)),
@@ -411,24 +422,58 @@ function epochNodeStream(ssb, groupId, opts = {}) {
 
   return deferredSource
 }
-function getGroupInit(ssb, groupId, cb) {
-  ssb.box2.getGroupInfo(groupId, (err, info) => {
-    // prettier-ignore
-    if (err) return cb(clarify(err, 'Failed to get group info for ' + groupId))
-    if (!info) return cb(new Error('Unknown group'))
 
-    // Fetch the tangle root
-    ssb.db.get(info.root, (err, rootVal) => {
-      // prettier-ignore
-      if (err) return cb(clarify(err, 'Failed to load group root with id ' + info.root))
+function getRootVal(ssb, msgId, cb) {
+  pull(
+    pullMany([
+      ssb.db.query(
+        where(and(isDecrypted('box2'), key(toMessageSigil(msgId)))),
+        live({ old: true }),
+        toPullStream()
+      ),
+      pull(
+        ssb.db.reindexed(),
+        pull.filter((msg) => fromMessageSigil(msg.key) === msgId)
+      ),
+    ]),
+    pull.take(1),
+    pull.drain(
+      (msg) => cb(null, msg.value),
+      (err) => {
+        if (err) cb(Error('Failed getting root msg async', { cause: err }))
+      }
+    )
+  )
+}
 
-      if (!isInitRoot(rootVal))
+function getGroupInit(ssb, groupId, cb) {
+  pull(
+    ssb.box2.getGroupInfoUpdates(groupId),
+    pull.take(1),
+    pull.drain(
+      (info) => {
+        if (!info) return cb(new Error('Unknown group'))
+
+        // Fetch the tangle root
+        // This is based on a live stream since sometimes the group info comes in very quick, before the root msg has had time to get put into the db
+        // and sometimes it might take ages (we haven't gotten that feed yet)
+        getRootVal(ssb, info.root, (err, rootVal) => {
+          // prettier-ignore
+          if (err) return cb(clarify(err, 'Failed to load group root with id ' + info.root))
+
+          if (!isInitRoot(rootVal))
+            // prettier-ignore
+            return cb(clarify(new Error(isInitRoot.string), 'Malformed group/init root message'))
+
+          cb(null, { key: info.root, value: rootVal })
+        })
+      },
+      (err) => {
         // prettier-ignore
-        return cb(clarify(new Error(isInitRoot.string), 'Malformed group/init root message'))
-
-      cb(null, { key: info.root, value: rootVal })
-    })
-  })
+        if (err) return cb(clarify(err, 'Failed to get group info for ' + groupId))
+      }
+    )
+  )
 }
 
 /* HELPERS */

lib/operators.js

@@ -5,7 +5,6 @@
 const { allocAndEncode, seekKey2, iterate } = require('bipf')
 const { equal } = require('jitdb/operators')
 
-const B_VALUE = allocAndEncode('value')
 const B_CONTENT = allocAndEncode('content')
 const B_TANGLES = allocAndEncode('tangles')
 const B_ROOT = allocAndEncode('root')
@@ -21,17 +20,15 @@ module.exports = {
     }
     const B_TANGLE = B_TANGLE_MAP.get(tangle)
 
-    function seekTangleRoot(buffer, start, p) {
-      if (p < 0) return -1
-      p = seekKey2(buffer, 0, B_VALUE, 0)
-      if (p < 0) return -1
-      p = seekKey2(buffer, p, B_CONTENT, 0)
-      if (p < 0) return -1
-      p = seekKey2(buffer, p, B_TANGLES, 0)
-      if (p < 0) return -1
-      p = seekKey2(buffer, p, B_TANGLE, 0)
-      if (p < 0) return -1
-      return seekKey2(buffer, p, B_ROOT, 0)
+    function seekTangleRoot(buffer, start, pValue) {
+      if (pValue < 0) return -1
+      const pValueContent = seekKey2(buffer, pValue, B_CONTENT, 0)
+      if (pValueContent < 0) return -1
+      const pValueContentTangles = seekKey2(buffer, pValueContent, B_TANGLES, 0)
+      if (pValueContentTangles < 0) return -1
+      const pValueContentTanglesTangle = seekKey2(buffer, pValueContentTangles, B_TANGLE, 0)
+      if (pValueContentTanglesTangle < 0) return -1
+      return seekKey2(buffer, pValueContentTanglesTangle, B_ROOT, 0)
     }
 
     return equal(seekTangleRoot, value, {
@@ -47,18 +44,18 @@ module.exports = {
   },
 }
 
-function seekFirstRecp(buffer, start, p) {
-  if (p < 0) return -1
-  p = seekKey2(buffer, p, B_CONTENT, 0)
-  if (p < 0) return -1
-  p = seekKey2(buffer, p, B_RECPS, 0)
-  if (p < 0) return -1
+function seekFirstRecp(buffer, start, pValue) {
+  if (pValue < 0) return -1
+  const pValueContent = seekKey2(buffer, pValue, B_CONTENT, 0)
+  if (pValueContent < 0) return -1
+  const pValueContentRecps = seekKey2(buffer, pValueContent, B_RECPS, 0)
+  if (pValueContentRecps < 0) return -1
 
-  let pValueFirstRecp
-  const error = iterate(buffer, p, (_, pointer) => {
-    pValueFirstRecp = pointer
+  let pValueContentRecps0
+  const error = iterate(buffer, pValueContentRecps, (_, pointer) => {
+    pValueContentRecps0 = pointer
     return true
   })
   if (error === -1) return -1
-  return pValueFirstRecp
+  return pValueContentRecps0
 }

package.json

@@ -1,6 +1,6 @@
 {
   "name": "ssb-tribes2",
-  "version": "1.3.0",
+  "version": "1.4.0",
   "description": "SSB private groups with ssb-db2",
   "repository": {
     "type": "git",

test/lib/epochs.test.js

@@ -26,7 +26,7 @@ test('lib/epochs (getEpochs, getMembers)', async (t) => {
   async function sync(label) {
     return run(`(sync ${label})`, replicate(peers), { isTest: false })
   }
-  t.teardown(() => peers.forEach((peer) => peer.close(true)))
+  t.teardown(() => Promise.all(peers.map((peer) => p(peer.close)(true))))
 
   const [aliceId, bobId, oscarId] = await getRootIds(peers)
   await run(
@@ -153,7 +153,7 @@ test('lib/epochs (getMissingMembers)', async (t) => {
       { isTest: false }
     )
   }
-  t.teardown(() => peers.forEach((peer) => peer.close(true)))
+  t.teardown(() => Promise.all(peers.map((peer) => p(peer.close)(true))))
 
   await run(
     'start tribes',
@@ -260,7 +260,7 @@ test('lib/epochs (getPreferredEpoch - 4.4. same membership)', async (t) => {
     Server({ name: 'bob' }),
     Server({ name: 'oscar' }),
   ]
-  t.teardown(() => peers.forEach((peer) => peer.close(true)))
+  t.teardown(() => Promise.all(peers.map((peer) => p(peer.close)(true))))
 
   const [alice, bob, oscar] = peers
   const [bobId, oscarId] = await getRootIds([bob, oscar])
@@ -381,7 +381,7 @@ test('lib/epochs (getPreferredEpoch - 4.5. subset membership)', async (t) => {
     Server({ name: 'carol' }),
     Server({ name: 'oscar' }),
   ]
-  t.teardown(() => peers.forEach((peer) => peer.close(true)))
+  t.teardown(() => Promise.all(peers.map((peer) => p(peer.close)(true))))
 
   const [alice, bob, carol, oscar] = peers
   const [bobId, carolId, oscarId] = await getRootIds([bob, carol, oscar])