feat: graft/prune events and mesh peer tagging

src/index.ts

@@ -179,6 +179,11 @@
    * Limits to bound protobuf decoding
    */
   decodeRpcLimits?: DecodeRPCLimits
+
+  /**
+   * If true, will utilize the libp2p connection manager tagging system to prune/graft connections to peers, defaults to false
+   */
+  taggingEnabled?: boolean
 }
 
 export interface GossipsubMessage {
@@ -1466,7 +1471,7 @@
     const now = Date.now()
     let doPX = this.opts.doPX
 
-    graft.forEach(({ topicID }) => {
+    graft.forEach(async ({ topicID }) => {
       if (!topicID) {
         return
       }
@@ -1540,6 +1545,16 @@
       peersInMesh.add(id)
 
       this.metrics?.onAddToMesh(topicID, InclusionReason.Subscribed, 1)
+
+      if (this.opts?.taggingEnabled ?? false) {
+        await this.components.peerStore.merge(peerIdFromString(id), {
+          tags: {
+            [topicID]: {
+              value: 100 // value should be 0-100
+            }
+          }
+        })
+      }
     })
 
     if (!prune.length) {
@@ -1594,6 +1609,14 @@
         }
         await this.pxConnect(peers)
       }
+
+      if (this.opts?.taggingEnabled ?? false) {
+        await this.components.peerStore.save(peerIdFromString(id), {
+          tags: {
+            [topicID]: {}
+          }
+        })
+      }
     }
   }
 
@@ -1832,9 +1855,9 @@
 
     this.mesh.set(topic, toAdd)
 
-    toAdd.forEach((id) => {
+    toAdd.forEach(async (id) => {
       this.log('JOIN: Add mesh link to %s in %s', id, topic)
-      this.sendGraft(id, topic)
+      await this.sendGraft(id, topic)
 
       // rust-libp2p
       // - peer_score.graft()
@@ -2190,14 +2213,24 @@
   /**
    * Sends a GRAFT message to a peer
    */
-  private sendGraft(id: PeerIdStr, topic: string): void {
+  private async sendGraft(id: PeerIdStr, topic: string): Promise<void> {
     const graft = [
       {
         topicID: topic
       }
     ]
 
     this.sendRpc(id, { control: { graft } })
+
+    if (this.opts?.taggingEnabled ?? false) {
+      await this.components.peerStore.merge(peerIdFromString(id), {
+        tags: {
+          [topic]: {
+            value: 100 // value should be 0-100
+          }
+        }
+      })
+    }
   }
 
   /**
@@ -2209,6 +2242,14 @@
     const prune = [await this.makePrune(id, topic, this.opts.doPX, onUnsubscribe)]
 
     this.sendRpc(id, { control: { prune } })
+
+    if (this.opts.taggingEnabled ?? false) {
+      await this.components.peerStore.save(peerIdFromString(id), {
+        tags: {
+          [topic]: {}
+        }
+      })
+    }
   }
 
   /**
@@ -2459,7 +2500,7 @@
 
         try {
           peerInfo = await this.components.peerStore.get(id)
         } catch (err: any) {
           if (err.code !== 'ERR_NOT_FOUND') {
             throw err
           }