feat: bootstrap and pubsub replicators

package.json

@@ -37,6 +37,7 @@
   },
   "prettier": "prettier-config-standard",
   "scripts": {
+    "generate": "protons src/message/*.proto",
     "prepublishOnly": "npm run build",
     "reset": "rm -rf node_modules package-lock.json && npm install",
     "lint": "aegir lint",
@@ -48,7 +49,7 @@
     "build": "aegir build && tsc-alias && ln -sf ./dist/test/.aegir.js ./.aegir.js",
     "docs": "NODE_OPTIONS=--max_old_space_size=4096 aegir docs",
     "test:chrome": "aegir test -b -t browser -f ./dist/test/**/test-*.js",
-    "test:node": "npm run clean && aegir test -b -t node -f ./dist/test/**/test-*.js -- --exit && npm run clean",
+    "test:node": "npm run build && npm run clean && aegir test -b -t node -f ./dist/test/**/test-*.js -- --exit && npm run clean",
     "test": "npm run test:node && npm run test:chrome",
     "clean": "git clean -fq test/fixtures && git clean -fqX test/temp .polendina/ && git checkout -- test/fixtures"
   },
@@ -104,6 +105,7 @@
     "path-browserify": "^1.0.1",
     "prettier": "^2.7.1",
     "prettier-config-standard": "^5.0.0",
+    "protons": "^7.0.4",
     "tsc-alias": "^1.8.7",
     "typescript": "^5.1.6",
     "wherearewe": "^2.0.1"
@@ -116,14 +118,21 @@
     "@libp2p/interfaces": "^3.3.2",
     "@libp2p/peer-id": "^2.0.3",
     "@libp2p/peer-id-factory": "^2.0.3",
+    "@open-draft/deferred-promise": "^2.1.0",
     "@tabcat/zzzync": "^2.0.0",
     "datastore-core": "^9.1.1",
+    "fission-bloom-filters": "^1.7.1",
     "helia": "^1.1.2",
     "interface-blockstore": "5.2",
     "it-all": "^2.0.0",
+    "it-concat": "^3.0.1",
     "it-drain": "3.0",
+    "it-length-prefixed": "^9.0.1",
+    "it-pushable": "^3.2.1",
     "multiformats": "^11.0.1",
     "p-queue": "^7.3.0",
+    "protons-runtime": "^5.0.1",
+    "streaming-iterables": "^7.1.0",
     "uint8arrays": "^4.0.3",
     "w3name": "^1.0.8",
     "web3.storage": "^4.5.4"

src/index.ts

@@ -10,6 +10,8 @@ export { Entry, basalEntry } from '@/entry/basal/index.js'
 export { Identity, basalIdentity } from '@/identity/basal/index.js'
 export { Keyvalue, keyvalueStore } from '@/store/keyvalue/index.js'
 export { liveReplicator } from '@/replicator/live/index.js'
+export { pubsubReplicator } from '@/replicator/pubsub/index.js'
+export { bootstrapReplicator } from '@/replicator/bootstrap/index.js'
 export { Address } from '@/manifest/index.js'
 export type { Manifest } from '@/manifest/index.js'
 export type { Playable } from '@/utils/playable.js'

src/message/heads.proto

@@ -0,0 +1,14 @@
+syntax = "proto3";
+
+message Message {
+  message Filter {
+    bytes data = 1;
+    uint32 hashes = 2;
+    optional uint32 seed = 3;
+  }
+
+  repeated bytes heads = 1;
+  optional Filter filter = 2;
+  optional bytes hash = 3;
+  optional bool match = 4;
+}

src/message/heads.ts

@@ -0,0 +1,173 @@
+/* eslint-disable import/export */
+/* eslint-disable complexity */
+/* eslint-disable @typescript-eslint/no-namespace */
+/* eslint-disable @typescript-eslint/no-unnecessary-boolean-literal-compare */
+/* eslint-disable @typescript-eslint/no-empty-interface */
+
+import { encodeMessage, decodeMessage, message } from 'protons-runtime'
+import type { Codec } from 'protons-runtime'
+import type { Uint8ArrayList } from 'uint8arraylist'
+
+export interface Message {
+  heads: Uint8Array[]
+  filter?: Message.Filter
+  hash?: Uint8Array
+  match?: boolean
+}
+
+export namespace Message {
+  export interface Filter {
+    data: Uint8Array
+    hashes: number
+    seed?: number
+  }
+
+  export namespace Filter {
+    let _codec: Codec<Filter>
+
+    export const codec = (): Codec<Filter> => {
+      if (_codec == null) {
+        _codec = message<Filter>((obj, w, opts = {}) => {
+          if (opts.lengthDelimited !== false) {
+            w.fork()
+          }
+
+          if ((obj.data != null && obj.data.byteLength > 0)) {
+            w.uint32(10)
+            w.bytes(obj.data)
+          }
+
+          if ((obj.hashes != null && obj.hashes !== 0)) {
+            w.uint32(16)
+            w.uint32(obj.hashes)
+          }
+
+          if (obj.seed != null) {
+            w.uint32(24)
+            w.uint32(obj.seed)
+          }
+
+          if (opts.lengthDelimited !== false) {
+            w.ldelim()
+          }
+        }, (reader, length) => {
+          const obj: any = {
+            data: new Uint8Array(0),
+            hashes: 0
+          }
+
+          const end = length == null ? reader.len : reader.pos + length
+
+          while (reader.pos < end) {
+            const tag = reader.uint32()
+
+            switch (tag >>> 3) {
+              case 1:
+                obj.data = reader.bytes()
+                break
+              case 2:
+                obj.hashes = reader.uint32()
+                break
+              case 3:
+                obj.seed = reader.uint32()
+                break
+              default:
+                reader.skipType(tag & 7)
+                break
+            }
+          }
+
+          return obj
+        })
+      }
+
+      return _codec
+    }
+
+    export const encode = (obj: Partial<Filter>): Uint8Array => {
+      return encodeMessage(obj, Filter.codec())
+    }
+
+    export const decode = (buf: Uint8Array | Uint8ArrayList): Filter => {
+      return decodeMessage(buf, Filter.codec())
+    }
+  }
+
+  let _codec: Codec<Message>
+
+  export const codec = (): Codec<Message> => {
+    if (_codec == null) {
+      _codec = message<Message>((obj, w, opts = {}) => {
+        if (opts.lengthDelimited !== false) {
+          w.fork()
+        }
+
+        if (obj.heads != null) {
+          for (const value of obj.heads) {
+            w.uint32(10)
+            w.bytes(value)
+          }
+        }
+
+        if (obj.filter != null) {
+          w.uint32(18)
+          Message.Filter.codec().encode(obj.filter, w)
+        }
+
+        if (obj.hash != null) {
+          w.uint32(26)
+          w.bytes(obj.hash)
+        }
+
+        if (obj.match != null) {
+          w.uint32(32)
+          w.bool(obj.match)
+        }
+
+        if (opts.lengthDelimited !== false) {
+          w.ldelim()
+        }
+      }, (reader, length) => {
+        const obj: any = {
+          heads: []
+        }
+
+        const end = length == null ? reader.len : reader.pos + length
+
+        while (reader.pos < end) {
+          const tag = reader.uint32()
+
+          switch (tag >>> 3) {
+            case 1:
+              obj.heads.push(reader.bytes())
+              break
+            case 2:
+              obj.filter = Message.Filter.codec().decode(reader, reader.uint32())
+              break
+            case 3:
+              obj.hash = reader.bytes()
+              break
+            case 4:
+              obj.match = reader.bool()
+              break
+            default:
+              reader.skipType(tag & 7)
+              break
+          }
+        }
+
+        return obj
+      })
+    }
+
+    return _codec
+  }
+
+  export const encode = (obj: Partial<Message>): Uint8Array => {
+    return encodeMessage(obj, Message.codec())
+  }
+
+  export const decode = (buf: Uint8Array | Uint8ArrayList): Message => {
+    return decodeMessage(buf, Message.codec())
+  }
+}

src/replica/index.ts

@@ -31,8 +31,8 @@ import { decodeCbor, encodeCbor } from '@/utils/block.js'
 
 const rootHashKey = new Key('rootHash')
 
-interface ReplicaEvents {
-  write: CustomEvent<undefined>
+export interface ReplicaEvents {
+  write: CustomEvent<EntryInstance<any>>
   update: CustomEvent<undefined>
 }
 
@@ -44,7 +44,7 @@ export class Replica extends Playable {
   readonly components: Pick<DbComponents, 'entry' | 'identity'>
 
   #datastore: Datastore
-  #blockstore: Blockstore
+  blockstore: Blockstore
   #graph: Graph | null
 
   _queue: PQueue
@@ -69,10 +69,10 @@ export class Replica extends Playable {
     const starting = async (): Promise<void> => {
       const root: BlockView<GraphRoot> | null = await getRoot(
         this.#datastore,
-        this.#blockstore
+        this.blockstore
       ).catch(() => null)
 
-      this.#graph = new Graph(this.#blockstore, root?.value)
+      this.#graph = new Graph(this.blockstore, root?.value)
       await start(this.#graph)
 
       if (root?.cid == null) {
@@ -95,7 +95,7 @@ export class Replica extends Playable {
     this.components = components
 
     this.#datastore = datastore
-    this.#blockstore = blockstore
+    this.blockstore = blockstore
     this.#graph = null
     this._queue = new PQueue({})
 
@@ -133,7 +133,7 @@ export class Replica extends Playable {
       direction: 'descend'
     }
   ): Promise<Array<EntryInstance<any>>> {
-    const blockstore = this.#blockstore
+    const blockstore = this.blockstore
     const entry = this.components.entry
     const identity = this.components.identity
 
@@ -180,7 +180,7 @@ export class Replica extends Playable {
   }
 
   async add (entries: Array<EntryInstance<any>>): Promise<void> {
-    if (this.#datastore == null || this.#blockstore == null) {
+    if (this.#datastore == null || this.blockstore == null) {
       throw new Error('replica not started')
     }
 
@@ -193,8 +193,8 @@ export class Replica extends Playable {
         continue
       }
 
-      await this.#blockstore.put(entry.cid, entry.block.bytes)
-      await this.#blockstore.put(entry.identity.auth, entry.identity.block.bytes)
+      await this.blockstore.put(entry.cid, entry.block.bytes)
+      await this.blockstore.put(entry.identity.auth, entry.identity.block.bytes)
 
       if (await this.access.canAppend(entry)) {
         await this.graph.add(entry.cid, entry.next)
@@ -221,10 +221,10 @@ export class Replica extends Playable {
       refs: [] // refs are empty for now
     })
 
-    await this.#blockstore.put(entry.cid, entry.block.bytes)
+    await this.blockstore.put(entry.cid, entry.block.bytes)
 
     return await this.add([entry]).then(() => {
-      this.events.dispatchEvent(new CustomEvent<undefined>('write'))
+      this.events.dispatchEvent(new CustomEvent<EntryInstance<any>>('write', { detail: entry }))
       return entry
     })
   }
@@ -278,7 +278,7 @@ export class Replica extends Playable {
 
   async #updateRoot (): Promise<void> {
     const block = await encodeRoot(this.graph.root)
-    await setRoot(this.#datastore, this.#blockstore, block)
+    await setRoot(this.#datastore, this.blockstore, block)
     this.root = block.cid
     this.events.dispatchEvent(new CustomEvent<undefined>('update'))
   }