Make all APIs async, internally wait for rescanning (#29)

lib/index.js

@@ -4,6 +4,8 @@ const b4a = require('b4a')
 const base58 = require('bs58')
 const Obz = require('obz')
 const Keypair = require('ppppp-keypair')
+const pull = require('pull-stream')
+const p = require('node:util').promisify
 const Log = require('./log')
 const MsgV4 = require('./msg-v4')
 const {
@@ -92,21 +94,31 @@ function assertValidConfig(config) {
 }
 
 class DBTangle extends MsgV4.Tangle {
-  /** @type {(msgID: MsgID) => Msg | undefined} */
+  /** @type {(msgID: MsgID, cb: CB<Msg>) => void} */
   #getMsg
 
   /**
    * @param {MsgID} rootID
-   * @param {Iterable<Rec>} recordsIter
+   * @param {(msgID: MsgID, cb: CB<Msg>) => void} getMsg
-   * @param {(msgID: MsgID) => Msg | undefined} getMsg
    */
-  constructor(rootID, recordsIter, getMsg) {
+  constructor(rootID, getMsg) {
     super(rootID)
     this.#getMsg = getMsg
-    for (const rec of recordsIter) {
-      if (!rec.msg) continue
-      this.add(rec.id, rec.msg)
   }
+
+  /**
+   * @param {MsgID} rootID
+   * @param {AsyncIterable<Rec>} recordsIter
+   * @param {(msgID: MsgID, cb: any) => void} getMsg
+   * @return {Promise<DBTangle>}
+   */
+  static async init(rootID, recordsIter, getMsg) {
+    const dbtangle = new DBTangle(rootID, getMsg)
+    for await (const rec of recordsIter) {
+      if (!rec.msg) continue
+      dbtangle.add(rec.id, rec.msg)
+    }
+    return dbtangle
   }
 
   /**
@@ -148,9 +160,13 @@ class DBTangle extends MsgV4.Tangle {
   /**
    * @param {Array<string>=} minSet
    * @param {Array<string>=} maxSet
-   * @returns {Array<Msg>}
+   * @param {CB<Array<Msg>>=} cb
+   * @return {Promise<Array<Msg>>|void}
    */
-  slice(minSet = [], maxSet = []) {
+  slice(minSet = [], maxSet = [], cb) {
+    // @ts-ignore
+    if (cb === undefined) return p(this.slice).bind(this)(minSet, maxSet)
+
     const minSetGood = minSet.filter((msgID) => this.has(msgID))
     const maxSetGood = maxSet.filter((msgID) => this.has(msgID))
     const minSetTight = this.getMinimumAmong(minSetGood)
@@ -164,21 +180,38 @@ class DBTangle extends MsgV4.Tangle {
     }
 
     const msgs = /**@type {Array<Msg>}*/ ([])
-    for (const msgID of this.topoSort()) {
+
-      if (trail.has(msgID)) {
+    pull(
-        const msg = this.#getMsg(msgID)
+      pull.values(this.topoSort()),
-        if (msg) msgs.push({ ...msg, data: null })
+      pull.asyncMap((msgID, cb) => {
+        this.#getMsg(msgID, (err, msg) => {
+          if (err) return cb(err)
+          cb(null, { id: msgID, msg })
+        })
+      }),
+      pull.drain(
+        (rec) => {
+          if (trail.has(rec.id)) {
+            if (rec.msg) msgs.push({ ...rec.msg, data: null })
           }
-      const isMin = minSetGood.includes(msgID)
+          const isMin = minSetGood.includes(rec.id)
-      const isMax = maxSetGood.includes(msgID)
+          const isMax = maxSetGood.includes(rec.id)
-      const isBeforeMin = minSetGood.some((min) => this.precedes(msgID, min))
+          const isBeforeMin = minSetGood.some((min) =>
-      const isAfterMax = maxSetGood.some((max) => this.precedes(max, msgID))
+            this.precedes(rec.id, min)
-      if (!isMin && isBeforeMin) continue
+          )
-      if (!isMax && isAfterMax) continue
+          const isAfterMax = maxSetGood.some((max) =>
-      const msg = this.#getMsg(msgID)
+            this.precedes(max, rec.id)
-      if (msg) msgs.push(msg)
+          )
+          if (!isMin && isBeforeMin) return
+          if (!isMax && isAfterMax) return
+          if (rec.msg) msgs.push(rec.msg)
+        },
+        (err) => {
+          if (err) return cb(Error('DBTangle.slice() failed', { cause: err }))
+          return cb(null, msgs)
         }
-    return msgs
+      )
+    )
   }
 }
 
@@ -272,17 +305,16 @@ function initDB(peer, config) {
     )
   })
 
+  let rescanning = /** @type {Doneable<null>} */ new Doneable()
+  rescanning.done()
+
   /**
-   * TODO: To fix. Notice that some synchronous read APIs such as `db.get()`,
-   * `db.msgs()`, `db.getTangle()` etc may read an *inconsistent* state of the
-   * `recs` array while rescanning is in progress. This may mean duplicate msgs
-   * are read. One possible fix for this is to make all public APIs async.
-   *
    * @param {CB<void>} cb
    */
   function rescanLogPostCompaction(cb) {
     miscRegistry = new WeakMap()
     let seq = -1
+    rescanning = new Doneable()
     log.scan(
       function rescanEach(offset, recInLog, size) {
         seq += 1
@@ -299,6 +331,7 @@ function initDB(peer, config) {
         // prettier-ignore
         if (err) return cb(new Error('Failed to rescan the log after compaction', { cause: err }))
         recs.length = seq + 1
+        rescanning.done()
         cb()
       },
       false // asRaw
@@ -364,33 +397,50 @@ function initDB(peer, config) {
 
   /**
    * @param {Array<MsgID>} tangleIDs
+   * @param {CB<Record<MsgID, DBTangle>>} cb
    */
-  function populateTangles(tangleIDs) {
+  function populateTangles(tangleIDs, cb) {
     /** @type {Record<MsgID, DBTangle>} */
     const tangles = {}
-    for (const tangleID of tangleIDs) {
+    pull(
-      tangles[tangleID] ??= new DBTangle(tangleID, records(), get)
+      pull.values(tangleIDs),
+      pull.asyncMap((tangleID, cb) => {
+        DBTangle.init(tangleID, records(), get).then((dbtangle) => {
+          tangles[tangleID] ??= dbtangle
+          cb(null, null)
+        })
+      }),
+      pull.drain(
+        () => {},
+        () => {
+          return cb(null, tangles)
         }
-    return tangles
+      )
+    )
   }
 
   /**
    * @param {Pick<RecPresent, 'id' | 'msg'>} rec
-   * @returns {DBTangle | null}
+   * @param {(err: Error | null, tangle: DBTangle | null) => void} cb
    */
-  function getAccountTangle(rec) {
+  function getAccountTangle(rec, cb) {
     const accountID = getAccountID(rec)
-    let accountTangle = /** @type {DBTangle | null} */ (null)
     if (accountID) {
-      accountTangle = new DBTangle(accountID, records(), get)
+      DBTangle.init(accountID, records(), get).then((accountTangle) => {
         if (rec.id === accountID) {
           accountTangle.add(rec.id, rec.msg)
         }
         if (!accountTangle.has(accountID)) {
-        throw new Error(`Account tangle "${accountID}" is locally unknown`)
+          return cb(
+            Error(`Account tangle "${accountID}" is locally unknown`),
+            null
+          )
         }
+        return cb(null, accountTangle)
+      })
+    } else {
+      return cb(null, null)
     }
-    return accountTangle
   }
 
   /**
@@ -399,16 +449,17 @@ function initDB(peer, config) {
    * @private
    * @param {DBTangle | null} accountTangle
    * @param {Msg['metadata']['accountTips']} accountTipsInMsg
-   * @returns {Set<string>}
+   * @param {CB<Set<string>>} cb
    */
-  function getSigkeysInAccount(accountTangle, accountTipsInMsg) {
+  function getSigkeysInAccount(accountTangle, accountTipsInMsg, cb) {
     const sigkeys = new Set()
-    if (!accountTangle) return sigkeys
+    if (!accountTangle) return cb(null, sigkeys)
 
-    const prunedTangle = accountTangle.slice(
+    accountTangle.slice(
       undefined,
-      accountTipsInMsg ?? undefined
+      accountTipsInMsg ?? undefined,
-    )
+      (err, prunedTangle) => {
+        if (err) return cb(err)
 
         for (const msg of prunedTangle) {
           if (!msg?.data) continue
@@ -426,7 +477,9 @@ function initDB(peer, config) {
             sigkeys.delete(data.key.bytes)
           }
         }
-    return sigkeys
+        return cb(null, sigkeys)
+      }
+    )
   }
 
   /**
@@ -434,7 +487,7 @@ function initDB(peer, config) {
    */
   function loaded(cb) {
     if (cb === void 0) return promisify(loaded)()
-    scannedLog.onDone(() => {
+    return scannedLog.onDone(() => {
       ghosts.onReady(cb)
     })
   }
@@ -446,55 +499,51 @@ function initDB(peer, config) {
    *
    * @param {Pick<RecPresent, 'id' | 'msg'>} rec
    * @param {MsgID} tangleID
-   * @returns {Error | null}
+   * @param {(err: Error | null, val: null) => void} cb
    */
-  function verifyRec(rec, tangleID) {
+  function verifyRec(rec, tangleID, cb) {
     let err
     // TODO: optimize this. This may be slow if you're adding many msgs in a
     // row, because it creates a new Map() each time. Perhaps with QuickLRU
-    const tangle = new DBTangle(tangleID, records(), get)
+    DBTangle.init(tangleID, records(), get).then((tangle) => {
       if (rec.id === tangleID) {
         tangle.add(rec.id, rec.msg)
       }
 
       if (MsgV4.isMoot(rec.msg)) {
         const sigkeys = new Set()
-      if ((err = MsgV4.validate(rec.msg, tangle, sigkeys, rec.id, tangleID))) {
+        if (
-        return new Error('Invalid msg', { cause: err })
+          (err = MsgV4.validate(rec.msg, tangle, sigkeys, rec.id, tangleID))
+        ) {
+          return cb(Error('Invalid msg', { cause: err }), null)
         }
-      return null
+        return cb(null, null)
       }
 
       // Identify the account and its sigkeys:
-    /** @type {DBTangle | null} */
+      getAccountTangle(rec, (err, accountTangle) => {
-    let accountTangle
+        // prettier-ignore
-    try {
+        if (err) return cb(Error('Unknown account tangle owning this msg', { cause: err }), null)
-      accountTangle = getAccountTangle(rec)
+
-    } catch (err) {
+        getSigkeysInAccount(
-      return new Error('Unknown account tangle owning this msg', { cause: err })
-    }
-    const sigkeys = getSigkeysInAccount(
           accountTangle,
-      rec.msg.metadata.accountTips
+          rec.msg.metadata.accountTips,
-    )
+          (err, sigkeys) => {
+            if (err) return cb(err, null)
 
             // Don't accept ghosts to come back, unless they are trail msgs
             if (!!rec.msg.data && ghosts.read(tangleID).has(rec.id)) {
-      return new Error('Refusing a ghost msg to come back')
+              return cb(Error('Refusing a ghost msg to come back'), null)
             }
 
-    if ((err = MsgV4.validate(rec.msg, tangle, sigkeys, rec.id, tangleID))) {
+            if (
-      return new Error('Invalid msg', { cause: err })
+              (err = MsgV4.validate(rec.msg, tangle, sigkeys, rec.id, tangleID))
-    }
+            ) {
-
+              return cb(Error('Invalid msg', { cause: err }), null)
-    // Account tangle related validations
-    if (rec.msg.metadata.account === ACCOUNT_SELF) {
-      const validAccountTangle = /** @type {Tangle} */ (accountTangle)
-      if ((err = validateAccountMsg(rec.msg, validAccountTangle))) {
-        return new Error('Invalid account msg', { cause: err })
-      }
             }
 
+            /** @param {(err: Error | null, val: null) => void} cb */
+            function verifyInner(cb) {
               // Unwrap encrypted inner msg and verify it too
               if (typeof rec.msg.data === 'string') {
                 const recDecrypted = decrypt(rec, peer, config)
@@ -502,15 +551,36 @@ function initDB(peer, config) {
                   const innerMsg = /** @type {Msg} */ (recDecrypted.msg.data)
                   const innerMsgID = MsgV4.getMsgID(innerMsg)
                   const innerRec = { id: innerMsgID, msg: innerMsg }
-        try {
+
-          verifyRec(innerRec, innerMsgID)
+                  verifyRec(innerRec, innerMsgID, (err) => {
-        } catch (err) {
+                    // prettier-ignore
-          return new Error('Failed to verify inner msg', { cause: err })
+                    if (err) return cb(Error('Failed to verify inner msg', { cause: err }), null)
+
+                    return cb(null, null)
+                  })
+                } else {
+                  return cb(null, null)
                 }
+              } else {
+                return cb(null, null)
               }
             }
 
-    return null
+            // Account tangle related validations
+            if (rec.msg.metadata.account === ACCOUNT_SELF) {
+              const validAccountTangle = /** @type {Tangle} */ (accountTangle)
+              validateAccountMsg(rec.msg, validAccountTangle, (err) => {
+                if (err)
+                  return cb(Error('Invalid account msg', { cause: err }), null)
+                return verifyInner(cb)
+              })
+            } else {
+              return verifyInner(cb)
+            }
+          }
+        )
+      })
+    })
   }
 
   /**
@@ -558,27 +628,30 @@ function initDB(peer, config) {
 
     // TODO: optimize this. Perhaps have a Map() of msgID -> record
     // Or even better, a bloom filter. If you just want to answer no/perhaps.
-    let rec
     let maybePredelete = bypassPredelete
-    if ((rec = getRecord(msgID))) {
+    getRecord(msgID, (err, gotRec) => {
+      if (err) return cb(Error('getRecord() failed in add()', { cause: err }))
+      let rec
+      if (gotRec) {
         // If existing record is dataless but new is dataful, then delete
-      if (rec.msg.data === null && msg.data !== null) {
+        if (gotRec.msg.data === null && msg.data !== null) {
           maybePredelete = del
           rec = { msg, id: msgID }
         } else {
-        return cb(null, rec)
+          return cb(null, gotRec)
         }
       } else rec = { msg, id: msgID }
 
       const actualTangleID = tangleID ?? inferTangleID(rec)
 
-    let err
+      verifyRec(rec, actualTangleID, (err) => {
-    if ((err = verifyRec(rec, actualTangleID))) {
+        if (err) {
           return cb(new Error('add() failed to verify msg', { cause: err }))
         }
 
         maybePredelete(msgID, (err) => {
-      if (err) return cb(new Error('add() failed to predelete', { cause: err }))
+          if (err)
+            return cb(new Error('add() failed to predelete', { cause: err }))
           // The majority of cases don't have ghosts to be removed, but this
           // operation is silent and cheap if there are no ghosts.
           removeGhost(actualTangleID, msgID, (err) => {
@@ -597,14 +670,16 @@ function initDB(peer, config) {
             })
           })
         })
+      })
+    })
   }
 
   /**
    * @param {Msg} msg
    * @param {Tangle} accountTangle
-   * @returns {string | undefined}
+   * @param {(err: Error | null, val: null) => void} cb
    */
-  function validateAccountMsg(msg, accountTangle) {
+  function validateAccountMsg(msg, accountTangle, cb) {
     if (!MsgV4.isRoot(msg)) {
       /** @type {AccountData} */
       const data = msg.data
@@ -614,13 +689,22 @@ function initDB(peer, config) {
           curve: /** @type {const} */ ('ed25519'),
           public: msg.sigkey,
         }
-        const powers = getAccountPowers(accountTangle, keypair)
+        getAccountPowers(accountTangle, keypair, (err, powers) => {
+          if (err) return cb(err, null)
+
           if (!powers.has('add')) {
             // prettier-ignore
-          return `invalid account msg: sigkey "${msg.sigkey}" does not have "add" power`
+            return cb(Error(`invalid account msg: sigkey "${msg.sigkey}" does not have "add" power`), null)
           }
+
+          return cb(null, null)
+        })
+      } else {
+        return cb(null, null)
       }
       // TODO validate 'del'
+    } else {
+      return cb(null, null)
     }
   }
 
@@ -632,7 +716,9 @@ function initDB(peer, config) {
     const keypair = opts.keypair ?? config.global.keypair
     const { account, domain } = opts
 
-    const mootID = findMoot(account, domain)?.id
+    findMoot(account, domain, (err, m) => {
+      if (err) return cb(err)
+      const mootID = m?.id
       if (mootID) return cb(null, mootID)
 
       const moot = MsgV4.createMoot(account, domain, keypair)
@@ -641,6 +727,7 @@ function initDB(peer, config) {
         if (err) return cb(new Error('initializeFeed() failed to add root', { cause: err }));
         cb(null, rec.id)
       })
+    })
   }
 
   /**
@@ -713,24 +800,41 @@ function initDB(peer, config) {
    *   keypair?: KeypairPublicSlice;
    *   account: string;
    * }} opts
-   * @returns {boolean}
+   * @param {(err: Error | null, has: boolean | null) => void} cb
    */
-  function accountHas(opts) {
+  function accountHas(opts, cb) {
     const keypair = opts?.keypair ?? config.global.keypair
 
-    const accountTangle = new DBTangle(opts.account, records(), get)
+    DBTangle.init(opts.account, records(), get).then((accountTangle) => {
-    for (const msgID of accountTangle.topoSort()) {
+      pull(
-      const msg = get(msgID)
+        pull.values(accountTangle.topoSort()),
-      if (!msg?.data) continue
+        pull.asyncMap((msgID, cb) => {
+          get(msgID, (err, msg) => {
+            // prettier-ignore
+            if (err) return cb(Error("db.account.has() failed to get() account tangle message", { cause: err }), null)
+
+            if (!msg?.data) return cb(null, false)
             /** @type {AccountData} */
             const data = msg.data
-      if (data.action !== 'add') continue
+            if (data.action !== 'add') return cb(null, false)
-      if (data.key.algorithm !== keypair.curve) continue
+            if (data.key.algorithm !== keypair.curve) return cb(null, false)
             if (data.key.bytes === keypair.public) {
-        return true
+              return cb(null, true)
             }
-    }
+            return cb(null, false)
-    return false
+          })
+        }),
+        pull.collect((err, results) => {
+          // prettier-ignore
+          if (err) return cb(Error('db.account.has() failed to calculate', { cause: err }), null)
+
+          return cb(
+            null,
+            results.some((res) => res === true)
+          )
+        })
+      )
+    })
   }
 
   /**
@@ -793,28 +897,43 @@ function initDB(peer, config) {
     })
   }
 
+  //* @param {(err: Error | null, val: Set<any> | null) => void} cb
   /**
    * @param {Tangle} accountTangle
    * @param {KeypairPublicSlice} keypair
-   * @returns {Set<AccountPower>}
+   * @param {CB<Set<AccountPower>>} cb
    */
-  function getAccountPowers(accountTangle, keypair) {
+  function getAccountPowers(accountTangle, keypair, cb) {
     const powers = new Set()
-    for (const msgID of accountTangle.topoSort()) {
+    pull(
-      const msg = get(msgID)
+      pull.values(accountTangle.topoSort()),
-      if (!msg?.data) continue
+      pull.asyncMap((msgID, cb) => {
+        get(msgID, (err, msg) => {
+          // prettier-ignore
+          if (err) return cb(Error("getAccountPowers() failed to get() account tangle message", { cause: err }))
+          if (!msg?.data) return cb(null, null)
           /** @type {AccountData} */
           const data = msg.data
-      if (data.action !== 'add') continue
+          if (data.action !== 'add') return cb(null, null)
-      if (data.key.algorithm !== keypair.curve) continue
+          if (data.key.algorithm !== keypair.curve) return cb(null, null)
-      if (data.key.bytes !== keypair.public) continue
+          if (data.key.bytes !== keypair.public) return cb(null, null)
           if (data.powers) {
             for (const power of data.powers) {
               powers.add(power)
             }
           }
+          cb(null, null)
+        })
+      }),
+      pull.drain(
+        () => {},
+        (err) => {
+          if (err)
+            return cb(Error('Failed getting account powers', { cause: err }))
+          cb(null, powers)
         }
-    return powers
+      )
+    )
   }
 
   /**
@@ -889,14 +1008,14 @@ function initDB(peer, config) {
     }
 
     // Verify powers of the signingKeypair:
-    const accountTangle = new DBTangle(opts.account, records(), get)
+    DBTangle.init(opts.account, records(), get).then((accountTangle) => {
-    if (obeying) {
+      getAccountPowers(accountTangle, signingKeypair, (err, signingPowers) => {
-      const signingPowers = getAccountPowers(accountTangle, signingKeypair)
+        if (err) return cb(err)
-      if (!signingPowers.has('add')) {
+
+        if (obeying && !signingPowers.has('add')) {
           // prettier-ignore
           return cb(new Error('account.add() failed because the signing keypair does not have the "add" power'))
         }
-    }
 
         // Verify input powers for the addedKeypair:
         if (obeying && opts.powers) {
@@ -918,7 +1037,15 @@ function initDB(peer, config) {
           }
         }
 
-    const accountRoot = get(opts.account)
+        get(opts.account, (err, accountRoot) => {
+          // prettier-ginore
+          if (err)
+            return cb(
+              Error('account.add() failed to get() account root', {
+                cause: err,
+              })
+            )
+
           if (!accountRoot) {
             // prettier-ignore
             return cb(new Error(`account.add() failed because the account root "${opts.account}" is unknown`))
@@ -963,6 +1090,9 @@ function initDB(peer, config) {
             queueMicrotask(() => onRecordAdded.set(rec))
             cb(null, rec)
           })
+        })
+      })
+    })
   }
 
   /**
@@ -988,14 +1118,22 @@ function initDB(peer, config) {
     const signingKeypair = config.global.keypair
 
     // Verify powers of the signingKeypair:
-    const accountTangle = new DBTangle(opts.account, records(), get)
+    DBTangle.init(opts.account, records(), get).then((accountTangle) => {
-    const signingPowers = getAccountPowers(accountTangle, signingKeypair)
+      getAccountPowers(accountTangle, signingKeypair, (err, signingPowers) => {
+        if (err) return cb(err)
+
         if (!signingPowers.has('del')) {
           // prettier-ignore
           return cb(new Error('account.del() failed because the signing keypair does not have the "del" power'))
         }
 
-    const accountRoot = get(opts.account)
+        get(opts.account, (err, accountRoot) => {
+          if (err)
+            return cb(
+              Error('account.del() failed to get() account root', {
+                cause: err,
+              })
+            )
           if (!accountRoot) {
             // prettier-ignore
             return cb(new Error(`account.del() failed because the account root "${opts.account}" is unknown`))
@@ -1031,6 +1169,9 @@ function initDB(peer, config) {
             queueMicrotask(() => onRecordAdded.set(rec))
             cb(null, rec)
           })
+        })
+      })
+    })
   }
 
   /**
@@ -1081,8 +1222,9 @@ function initDB(peer, config) {
       // Fill-in tangle opts:
       const tangleTemplates = opts.tangles ?? []
       tangleTemplates.push(mootID)
-      const tangles = populateTangles(tangleTemplates)
+      populateTangles(tangleTemplates, (err, tangles) => {
-      const accountTangle = new DBTangle(opts.account, records(), get)
+        if (err) return cb(err)
+        DBTangle.init(opts.account, records(), get).then((accountTangle) => {
           const accountTips = [...accountTangle.tips]
           /**@type {MsgV4.CreateOpts}*/
           const fullOpts = { ...opts, tangles, accountTips, keypair }
@@ -1137,46 +1279,57 @@ function initDB(peer, config) {
             cb(null, rec)
           })
         })
+      })
+    })
   }
 
   /**
    * @param {string} id
    * @param {string} findDomain
-   * @returns {RecPresent | null}
+   * @param {(err: Error | null, rec: RecPresent | null) => void} cb
    */
-  function findMoot(id, findDomain) {
+  function findMoot(id, findDomain, cb) {
     const findAccount = MsgV4.stripAccount(id)
-    for (const rec of records()) {
+
+    new Promise(async (res) => {
+      for await (const rec of records()) {
         if (rec.msg && MsgV4.isMoot(rec.msg, findAccount, findDomain)) {
-        return rec
+          return res(rec)
         }
       }
-    return null
+      return res(null)
+    }).then((value) => cb(null, value))
   }
 
   /**
    * @param {MsgID} msgID
-   * @returns {RecPresent | null}
+   * @param {(err: Error | null, rec: RecPresent | null) => void} cb
    */
-  function getRecord(msgID) {
+  function getRecord(msgID, cb) {
     // TODO: improve performance of this when getting many messages, the arg
     // could be an array of hashes, so we can do a single pass over the records.
+    rescanning.onDone(() => {
       const isUri = msgID.startsWith('ppppp:')
       for (let i = 0; i < recs.length; i++) {
         const rec = recs[i]
         if (!rec) continue
-      if (isUri && rec.id && msgID.endsWith(rec.id)) return rec
+        if (isUri && rec.id && msgID.endsWith(rec.id)) return cb(null, rec)
-      else if (!isUri && rec.id === msgID) return rec
+        else if (!isUri && rec.id === msgID) return cb(null, rec)
       }
-    return null
+      return cb(null, null)
+    })
   }
 
   /**
    * @param {MsgID} msgID
-   * @returns {Msg | undefined}
+   * @param {(err: Error | null, msg?: Msg) => void} cb
    */
-  function get(msgID) {
+  function get(msgID, cb) {
-    return getRecord(msgID)?.msg
+    getRecord(msgID, (err, rec) => {
+      // prettier-ignore
+      if (err) return cb(Error("Failed to getRecord() when get()ting message", { cause: err }))
+      return cb(null, rec?.msg)
+    })
   }
 
   /**
@@ -1184,9 +1337,12 @@ function initDB(peer, config) {
    * @param {CB<void>} cb
    */
   function del(msgID, cb) {
-    const rec = getRecord(msgID)
+    getRecord(msgID, (err, rec) => {
+      // prettier-ignore
+      if (err) return cb(Error("Couldn't getRecord when del()eting it", { cause: err }))
       if (!rec) return cb()
       if (!rec.msg) return cb()
+      rescanning.onDone(() => {
         const misc = miscRegistry.get(rec)
         const seq = misc?.seq ?? -1
         const offset = misc?.offset ?? -1
@@ -1202,6 +1358,8 @@ function initDB(peer, config) {
             cb()
           })
         })
+      })
+    })
   }
 
   /**
@@ -1217,7 +1375,11 @@ function initDB(peer, config) {
     // prettier-ignore
     if (!opts.span || typeof opts.span !== 'number') return cb(new Error('ghosts.add() requires span in `opts.span`'))
     const { tangleID, msgID, span } = opts
-    const rec = getRecord(msgID)
+    getRecord(msgID, (err, rec) => {
+      if (err)
+        return cb(
+          Error('Failed to getRecord() in ghosts.add()', { cause: err })
+        )
       if (!rec) return cb()
       if (!rec.msg) return cb()
       const tangleData = rec.msg.metadata.tangles[tangleID]
@@ -1230,6 +1392,7 @@ function initDB(peer, config) {
         if (err) cb(new Error('ghosts.add() failed to save to disk', { cause: err }))
         else cb()
       })
+    })
   }
 
   /**
@@ -1277,7 +1440,8 @@ function initDB(peer, config) {
    * @param {CB<void>} cb
    */
   function erase(msgID, cb) {
-    const rec = getRecord(msgID)
+    getRecord(msgID, (err, rec) => {
+      if (err) return cb(Error('erase() failed to getRecord()', { cause: err }))
       if (!rec) return cb()
       if (!rec.msg) return cb()
       if (!rec.msg.data) return cb()
@@ -1297,29 +1461,35 @@ function initDB(peer, config) {
           cb()
         })
       })
+    })
   }
 
   /**
    * @param {MsgID} tangleID
-   * @returns {DBTangle | null}
+   * @param {(err: Error | null, tangle: DBTangle | null) => void} cb
    */
-  function getTangle(tangleID) {
+  function getTangle(tangleID, cb) {
-    const tangle = new DBTangle(tangleID, records(), get)
+    DBTangle.init(tangleID, records(), get).then((tangle) => {
       if (tangle.size > 0) {
-      return tangle
+        return cb(null, tangle)
       } else {
-      return null
+        return cb(null, null)
       }
+    })
   }
 
-  function* msgs() {
+  async function* msgs() {
+    await p(rescanning.onDone).bind(rescanning)()
+
     for (let i = 0; i < recs.length; i++) {
       const rec = recs[i]
       if (rec?.msg) yield rec.msg
     }
   }
 
-  function* records() {
+  async function* records() {
+    await p(rescanning.onDone).bind(rescanning)()
+
     for (let i = 0; i < recs.length; i++) {
       const rec = recs[i]
       if (rec) yield rec

lib/msg-v4/validation.js

@@ -334,7 +334,7 @@ function validate(msg, tangle, sigkeys, msgID, rootID) {
 
   try {
     if (tangle.type === 'feed' && isMoot(msg)) return // nothing else to check
-  } catch (err) {
+  } catch (/** @type {any} */ err) {
     return err
   }
 

package.json

@@ -29,8 +29,8 @@
   "dependencies": {
     "@alloc/quick-lru": "^5.2.0",
     "atomic-file-rw": "~0.3.0",
-    "blake3": "~2.1.7",
     "b4a": "~1.6.4",
+    "blake3": "~2.1.7",
     "bs58": "~5.0.0",
     "debug": "^4.3.0",
     "is-buffer-zero": "^1.0.0",
@@ -39,15 +39,17 @@
     "multicb": "~1.2.2",
     "mutexify": "~1.4.0",
     "obz": "~1.1.0",
-    "ppppp-keypair": "github:staltz/ppppp-keypair#61ef4420578f450dc2cc7b1efc1c5a691a871c74",
     "polyraf": "^1.1.0",
+    "ppppp-keypair": "github:staltz/ppppp-keypair#61ef4420578f450dc2cc7b1efc1c5a691a871c74",
     "promisify-4loc": "~1.0.0",
     "promisify-tuple": "~1.2.0",
+    "pull-stream": "^3.7.0",
     "push-stream": "~11.2.0",
     "set.prototype.union": "~1.0.2"
   },
   "devDependencies": {
     "@types/b4a": "^1.6.0",
+    "@types/pull-stream": "^3.6.7",
     "c8": "^7.11.0",
     "flumecodec": "~0.0.1",
     "husky": "^4.3.0",
@@ -55,8 +57,8 @@
     "prettier": "^2.6.2",
     "pretty-quick": "^3.1.3",
     "rimraf": "^4.4.0",
-    "secret-stack": "8.0.0",
     "secret-handshake-ext": "0.0.10",
+    "secret-stack": "8.0.0",
     "ssb-box": "^1.0.1",
     "typescript": "^5.1.3"
   },

test/account-add.test.js

@@ -24,7 +24,10 @@ test('account.add()', async (t) => {
       subdomain: 'person',
     })
 
-    assert.equal(peer.db.account.has({ account, keypair: keypair2 }), false)
+    assert.equal(
+      await p(peer.db.account.has)({ account, keypair: keypair2 }),
+      false
+    )
 
     const consent = peer.db.account.consent({ account, keypair: keypair2 })
 
@@ -61,7 +64,10 @@ test('account.add()', async (t) => {
     )
     assert.equal(msg.sigkey, keypair1.public, 'msg.sigkey OLD KEY')
 
-    assert.equal(peer.db.account.has({ account, keypair: keypair2 }), true)
+    assert.equal(
+      await p(peer.db.account.has)({ account, keypair: keypair2 }),
+      true
+    )
 
     await p(peer.close)()
   })
@@ -79,7 +85,7 @@ test('account.add()', async (t) => {
       keypair: keypair1,
       subdomain: 'account',
     })
-    const msg1 = peer1.db.get(id)
+    const msg1 = await p(peer1.db.get)(id)
 
     const { msg: msg2 } = await p(peer1.db.account.add)({
       account: id,
@@ -88,7 +94,10 @@ test('account.add()', async (t) => {
     })
     assert.equal(msg2.data.key.bytes, keypair2.public)
 
-    assert.equal(peer1.db.account.has({ account: id, keypair: keypair2 }), true)
+    assert.equal(
+      await p(peer1.db.account.has)({ account: id, keypair: keypair2 }),
+      true
+    )
 
     await p(peer1.close)()
     rimraf.sync(DIR)
@@ -151,7 +160,7 @@ test('account.add()', async (t) => {
       keypair: keypair1,
       subdomain: 'person',
     })
-    const accountMsg0 = peer.db.get(account)
+    const accountMsg0 = await p(peer.db.get)(account)
 
     // Consent is implicitly created because keypair2 has .private
     const accountRec1 = await p(peer.db.account.add)({
@@ -166,10 +175,13 @@ test('account.add()', async (t) => {
       keypair: keypair2,
     })
     assert.equal(postRec.msg.data.text, 'hello', 'post text correct')
-    const mootRec = peer.db.feed.findMoot(account, 'post')
+    const mootRec = await p(peer.db.feed.findMoot)(account, 'post')
     assert.ok(mootRec, 'posts moot exists')
 
-    const recs = [...peer.db.records()]
+    const recs = []
+    for await (rec of peer.db.records()) {
+      recs.push(rec)
+    }
     assert.equal(recs.length, 4, '4 records')
     const [_accountRec0, _accountRec1, postsRoot, _post] = recs
     assert.deepEqual(_accountRec0.msg, accountMsg0, 'accountMsg0')
@@ -224,7 +236,7 @@ test('account.add()', async (t) => {
         keypair: keypair1,
         subdomain: 'person',
       })
-      const accountMsg0 = peer.db.get(account)
+      const accountMsg0 = await p(peer.db.get)(account)
 
       const consent = peer.db.account.consent({ account, keypair: keypair2 })
 
@@ -241,7 +253,7 @@ test('account.add()', async (t) => {
         data: { text: 'potato' },
         keypair: keypair2,
       })
-      const postMootRec = peer.db.feed.findMoot(account, 'post')
+      const postMootRec = await p(peer.db.feed.findMoot)(account, 'post')
 
       const delRec = await p(peer.db.account.del)({
         account,

test/account-create.test.js

@@ -21,7 +21,7 @@ test('account.create() ', async (t) => {
       _nonce: 'MYNONCE',
     })
     assert.ok(account, 'accountRec0 exists')
-    const msg = peer.db.get(account)
+    const msg = await p(peer.db.get)(account)
     assert.deepEqual(
       msg.data,
       {
@@ -60,7 +60,7 @@ test('account.create() ', async (t) => {
       subdomain: 'person',
     })
     assert.ok(account, 'account created')
-    const msg = peer.db.get(account)
+    const msg = await p(peer.db.get)(account)
     assert.equal(msg.data.key.bytes, keypair.public, 'msg.data')
     assert.equal(msg.metadata.account, 'self', 'msg.metadata.account')
     assert.equal(msg.metadata.accountTips, null, 'msg.metadata.accountTips')
@@ -129,7 +129,7 @@ test('account.create() ', async (t) => {
       subdomain,
     })
     assert.ok(account, 'account created')
-    const msg = peer.db.get(account)
+    const msg = await p(peer.db.get)(account)
     assert.equal(msg.data.key.bytes, keypair.public, 'msg.data')
     assert.equal(msg.metadata.account, 'self', 'msg.metadata.account')
     assert.equal(msg.metadata.accountTips, null, 'msg.metadata.accountTips')

test/add.test.js

@@ -103,7 +103,7 @@ test('add()', async (t) => {
     {
       const ids = []
       const texts = []
-      for (const rec of peer.db.records()) {
+      for await (const rec of peer.db.records()) {
         if (rec.msg.metadata.domain === 'something') {
           ids.push(rec.id)
           texts.push(rec.msg.data?.text)
@@ -122,7 +122,7 @@ test('add()', async (t) => {
     {
       const ids = []
       const texts = []
-      for (const rec of peer.db.records()) {
+      for await (const rec of peer.db.records()) {
         if (rec.msg.metadata.domain === 'something') {
           ids.push(rec.id)
           texts.push(rec.msg.data?.text)

test/del.test.js

@@ -36,7 +36,7 @@ test('del()', async (t) => {
 
   {
     const texts = []
-    for (const msg of peer.db.msgs()) {
+    for await (const msg of peer.db.msgs()) {
       if (msg.data && msg.metadata.account?.length > 4) {
         texts.push(msg.data.text)
       }
@@ -60,7 +60,7 @@ test('del()', async (t) => {
 
   {
     const texts = []
-    for (const msg of peer.db.msgs()) {
+    for await (const msg of peer.db.msgs()) {
       if (msg.data && msg.metadata.account?.length > 4) {
         texts.push(msg.data.text)
       }
@@ -76,7 +76,7 @@ test('del()', async (t) => {
 
   {
     const texts = []
-    for (const msg of peer.db.msgs()) {
+    for await (const msg of peer.db.msgs()) {
       if (msg.data && msg.metadata.account?.length > 4) {
         texts.push(msg.data.text)
       }

test/erase.test.js

@@ -36,7 +36,7 @@ test('erase()', async (t) => {
   const SAVED_UPON_ERASE = '{"text":"m*"}'.length - 'null'.length
 
   const before = []
-  for (const msg of peer.db.msgs()) {
+  for await (const msg of peer.db.msgs()) {
     if (msg.data && msg.metadata.account?.length > 4) {
       before.push(msg.data.text)
     }
@@ -59,7 +59,7 @@ test('erase()', async (t) => {
   await p(peer.db.erase)(msgIDs[2])
 
   const after = []
-  for (const msg of peer.db.msgs()) {
+  for await (const msg of peer.db.msgs()) {
     if (msg.data && msg.metadata.account?.length > 4) {
       after.push(msg.data.text)
     }
@@ -68,7 +68,7 @@ test('erase()', async (t) => {
   assert.deepEqual(after, ['m0', 'm1', 'm3', 'm4'], '4 msgs after the erase')
 
   const after2 = []
-  for (const msg of peer.db.msgs()) {
+  for await (const msg of peer.db.msgs()) {
     for (const tangleID in msg.metadata.tangles) {
       after2.push(msg.metadata.tangles[tangleID].depth)
     }

test/feed-find-moot.test.js

@@ -23,7 +23,7 @@ test('feed.findMoot()', async (t) => {
 
   await p(peer.db.add)(moot, mootID)
 
-  const mootRec = peer.db.feed.findMoot(id, 'post')
+  const mootRec = await p(peer.db.feed.findMoot)(id, 'post')
   assert.equal(mootRec.id, mootID, 'feed.findMoot() returns moot ID')
 
   await p(peer.close)(true)

test/feed-publish.test.js

@@ -127,7 +127,7 @@ test('feed.publish()', async (t) => {
     assert.equal(typeof recEncrypted.msg.data, 'string')
     assert.ok(recEncrypted.msg.data.endsWith('.box'), '.box')
 
-    const msgDecrypted = peer.db.get(recEncrypted.id)
+    const msgDecrypted = await p(peer.db.get)(recEncrypted.id)
     assert.equal(msgDecrypted.data.text, 'I am chewing food')
   })
 

test/get.test.js

@@ -28,7 +28,7 @@ test('get()', async (t) => {
   })
   const msgID1 = MsgV4.getMsgID(rec1.msg)
 
-  const msg = peer.db.get(msgID1)
+  const msg = await p(peer.db.get)(msgID1)
   assert.ok(msg, 'msg exists')
   assert.equal(msg.data.text, 'I am 1st post')
 

test/getTangle.test.js

@@ -103,12 +103,12 @@ test('getTangle()', async (t) => {
     reply3LoText = reply3B.localeCompare(reply3C) < 0 ? 'reply 3B' : 'reply 3C'
     reply3HiText = reply3B.localeCompare(reply3C) < 0 ? 'reply 3C' : 'reply 3B'
 
-    tangle = peer.db.getTangle(rootPost)
+    tangle = await p(peer.db.getTangle)(rootPost)
   }
 
-  await t.test('getTangle unknown ID returns null', (t) => {
+  await t.test('getTangle unknown ID returns null', async (t) => {
     assert.equal(
-      peer.db.getTangle('Lq6xwbdvGVmSsY3oYRugpZ3DY8chX9SLhRhjJKyZHQn'),
+      await p(peer.db.getTangle)('Lq6xwbdvGVmSsY3oYRugpZ3DY8chX9SLhRhjJKyZHQn'),
       null
     )
   })
@@ -280,9 +280,9 @@ test('getTangle()', async (t) => {
     assert.deepEqual(actual4, expected4)
   })
 
-  await t.test('Tangle.slice', (t) => {
+  await t.test('Tangle.slice', async (t) => {
     {
-      const msgs = tangle.slice()
+      const msgs = await tangle.slice()
       const texts = msgs.map((msg) => msg.data?.text)
       assert.deepEqual(texts, [
         'root',
@@ -295,13 +295,13 @@ test('getTangle()', async (t) => {
     }
 
     {
-      const msgs = tangle.slice([], [reply2])
+      const msgs = await tangle.slice([], [reply2])
       const texts = msgs.map((msg) => msg.data?.text)
       assert.deepEqual(texts, ['root', reply1LoText, reply1HiText, 'reply 2'])
     }
 
     {
-      const msgs = tangle.slice([reply2], [])
+      const msgs = await tangle.slice([reply2], [])
       const texts = msgs.map((msg) => msg.data?.text)
       assert.deepEqual(texts, [
         undefined, // root
@@ -313,7 +313,7 @@ test('getTangle()', async (t) => {
     }
 
     {
-      const msgs = tangle.slice([reply2], [reply2])
+      const msgs = await tangle.slice([reply2], [reply2])
       const texts = msgs.map((msg) => msg.data?.text)
       assert.deepEqual(texts, [
         undefined, // root
@@ -323,7 +323,7 @@ test('getTangle()', async (t) => {
     }
 
     {
-      const msgs = tangle.slice([reply2], [reply2, reply3Lo])
+      const msgs = await tangle.slice([reply2], [reply2, reply3Lo])
       const texts = msgs.map((msg) => msg.data?.text)
       assert.deepEqual(texts, [
         undefined, // root
@@ -343,7 +343,7 @@ test('getTangle()', async (t) => {
       await p(peer.db.erase)(msgID)
     }
 
-    const tangle2 = peer.db.getTangle(rootPost)
+    const tangle2 = await p(peer.db.getTangle)(rootPost)
     const sorted = tangle2.topoSort()
 
     assert.deepEqual(sorted, [rootPost, reply3Lo, reply3Hi])

test/ghosts.tests.js

@@ -6,6 +6,7 @@ const p = require('node:util').promisify
 const rimraf = require('rimraf')
 const Keypair = require('ppppp-keypair')
 const { createPeer } = require('./util')
+const MsgV4 = require('../lib/msg-v4')
 
 const DIR = path.join(os.tmpdir(), 'ppppp-db-ghosts')
 rimraf.sync(DIR)
@@ -28,7 +29,7 @@ test('ghosts.add, ghosts.get, ghosts.getMinDepth', async (t) => {
     })
     msgIDs.push(rec.id)
   }
-  const tangleID = peer.db.feed.findMoot(account, 'post')?.id
+  const tangleID = (await p(peer.db.feed.findMoot)(account, 'post'))?.id
 
   const ghosts0 = peer.db.ghosts.get(tangleID)
   assert.deepEqual(ghosts0, [], 'no ghosts so far')
@@ -71,7 +72,8 @@ test('ghosts.add queues very-concurrent calls', async (t) => {
     })
     msgIDs.push(rec.id)
   }
-  const tangleID = peer.db.feed.findMoot(account, 'post')?.id
+  const moot = MsgV4.createMoot(account, 'post', keypair)
+  const tangleID = MsgV4.getMsgID(moot)
 
   const ghosts0 = peer.db.ghosts.get(tangleID)
   assert.deepEqual(ghosts0, [], 'no ghosts so far')

test/msgs-iterator.test.js

@@ -31,7 +31,7 @@ test('msgs() iterator', async (t) => {
 
   const posts = []
   const abouts = []
-  for (const msg of peer.db.msgs()) {
+  for await (const msg of peer.db.msgs()) {
     if (!msg.data) continue
     if (msg.metadata.domain === 'post') posts.push(msg.data.text)
     else if (msg.metadata.domain === 'about') abouts.push(msg.data.name)

test/re-open.test.js

@@ -41,7 +41,7 @@ test('publish some msgs, close, re-open', async (t) => {
   await peer2.db.loaded()
 
   const texts = []
-  for (const msg of peer2.db.msgs()) {
+  for await (const msg of peer2.db.msgs()) {
     if (!msg.data || !(msg.metadata.account?.length > 4)) continue
     texts.push(msg.data.text)
   }

test/records-iterator.test.js

@@ -29,7 +29,7 @@ test('records() iterator', async (t) => {
   }
 
   let count = 0
-  for (const rec of peer.db.records()) {
+  for await (const rec of peer.db.records()) {
     if (!rec.msg.data) continue
     if (rec.msg.metadata.account === 'self') continue
     assert.ok(rec.received, 'received')

test/sigkeys.test.js

@@ -31,7 +31,7 @@ test('sigkeys', async (t) => {
         keypair: keypair1,
         subdomain: 'person',
       })
-      const accountMsg0 = peer.db.get(account)
+      const accountMsg0 = await p(peer.db.get)(account)
 
       const consent = peer.db.account.consent({ account, keypair: keypair2 })
 
@@ -50,7 +50,7 @@ test('sigkeys', async (t) => {
       })
 
       const postMootId = peer.db.feed.getID(account, 'post')
-      const postMootMsg = peer.db.get(postMootId)
+      const postMootMsg = await p(peer.db.get)(postMootId)
 
       const tangle = new MsgV4.Tangle(postMootId)
       tangle.add(postMootId, postMootMsg)

tsconfig.json

@@ -1,17 +1,31 @@
 {
-  "include": ["declarations", "lib/**/*.js"],
+  "include": [
-  "exclude": ["coverage/", "node_modules/", "test/"],
+    "declarations",
+    "lib/**/*.js"
+  ],
+  "exclude": [
+    "coverage/",
+    "node_modules/",
+    "test/"
+  ],
   "compilerOptions": {
     "checkJs": true,
     "declaration": true,
     "emitDeclarationOnly": true,
     "exactOptionalPropertyTypes": true,
     "forceConsistentCasingInFileNames": true,
-    "lib": ["es2022", "dom"],
+    "noImplicitReturns": true,
+    "lib": [
+      "es2022",
+      "dom"
+    ],
     "module": "node16",
     "skipLibCheck": true,
     "strict": true,
     "target": "es2022",
-    "typeRoots": ["node_modules/@types", "declarations"]
+    "typeRoots": [
+      "node_modules/@types",
+      "declarations"
+    ]
   }
 }