feat: add consume peer record method (#84)

Add method to patch peer info with certified addresses extracted from a signed peer record.

package.json

@@ -151,6 +151,7 @@
     "@libp2p/interfaces": "^3.2.0",
     "@libp2p/logger": "^2.0.7",
     "@libp2p/peer-id": "^2.0.0",
+    "@libp2p/peer-record": "^5.0.3",
     "@multiformats/multiaddr": "^12.0.0",
     "interface-datastore": "^8.0.0",
     "mortice": "^3.0.1",

src/index.ts

@@ -6,6 +6,7 @@ import type { Datastore } from 'interface-datastore'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import type { Libp2pEvents } from '@libp2p/interface-libp2p'
 import { logger } from '@libp2p/logger'
+import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record'
 
 const log = logger('libp2p:peer-store')
 
@@ -164,6 +165,47 @@ export class PersistentPeerStore implements PeerStore {
     }
   }
 
+  async consumePeerRecord (buf: Uint8Array, expectedPeer?: PeerId): Promise<boolean> {
+    const envelope = await RecordEnvelope.openAndCertify(buf, PeerRecord.DOMAIN)
+
+    if (expectedPeer?.equals(envelope.peerId) === false) {
+      log('envelope peer id was not the expected peer id - expected: %p received: %p', expectedPeer, envelope.peerId)
+      return false
+    }
+
+    const peerRecord = PeerRecord.createFromProtobuf(envelope.payload)
+    let peer: Peer | undefined
+
+    try {
+      peer = await this.get(envelope.peerId)
+    } catch (err: any) {
+      if (err.code !== 'ERR_NOT_FOUND') {
+        throw err
+      }
+    }
+
+    // ensure seq is greater than, or equal to, the last received
+    if (peer?.peerRecordEnvelope != null) {
+      const storedEnvelope = await RecordEnvelope.createFromProtobuf(peer.peerRecordEnvelope)
+      const storedRecord = PeerRecord.createFromProtobuf(storedEnvelope.payload)
+
+      if (storedRecord.seqNumber >= peerRecord.seqNumber) {
+        log('sequence number was lower or equal to existing sequence number - stored: %d received: %d', storedRecord.seqNumber, peerRecord.seqNumber)
+        return false
+      }
+    }
+
+    await this.patch(peerRecord.peerId, {
+      peerRecordEnvelope: buf,
+      addresses: peerRecord.multiaddrs.map(multiaddr => ({
+        isCertified: true,
+        multiaddr
+      }))
+    })
+
+    return true
+  }
+
   #emitIfUpdated (id: PeerId, result: PeerUpdate): void {
     if (!result.updated) {
       return

test/index.spec.ts

@@ -10,6 +10,7 @@ import { createEd25519PeerId } from '@libp2p/peer-id-factory'
 import delay from 'delay'
 import { EventEmitter } from '@libp2p/interfaces/events'
 import type { Libp2pEvents } from '@libp2p/interface-libp2p'
+import { RecordEnvelope, PeerRecord } from '@libp2p/peer-record'
 
 const addr1 = multiaddr('/ip4/127.0.0.1/tcp/8000')
 
@@ -162,4 +163,101 @@ describe('PersistentPeerStore', () => {
         .that.does.not.have.key(name)
     })
   })
+
+  describe('peer record', () => {
+    it('consumes a peer record, creating a peer', async () => {
+      const peerRecord = new PeerRecord({
+        peerId,
+        multiaddrs: [
+          multiaddr('/ip4/127.0.0.1/tcp/1234')
+        ]
+      })
+      const signedPeerRecord = await RecordEnvelope.seal(peerRecord, peerId)
+
+      await expect(peerStore.has(peerId)).to.eventually.be.false()
+      await peerStore.consumePeerRecord(signedPeerRecord.marshal())
+      await expect(peerStore.has(peerId)).to.eventually.be.true()
+
+      const peer = await peerStore.get(peerId)
+      expect(peer.addresses.map(({ multiaddr, isCertified }) => ({
+        isCertified,
+        multiaddr: multiaddr.toString()
+      }))).to.deep.equal([{
+        isCertified: true,
+        multiaddr: '/ip4/127.0.0.1/tcp/1234'
+      }])
+    })
+
+    it('overwrites old addresses with those from a peer record', async () => {
+      await peerStore.patch(peerId, {
+        multiaddrs: [
+          multiaddr('/ip4/127.0.0.1/tcp/1234')
+        ]
+      })
+
+      const peerRecord = new PeerRecord({
+        peerId,
+        multiaddrs: [
+          multiaddr('/ip4/127.0.0.1/tcp/4567')
+        ]
+      })
+      const signedPeerRecord = await RecordEnvelope.seal(peerRecord, peerId)
+
+      await peerStore.consumePeerRecord(signedPeerRecord.marshal())
+
+      await expect(peerStore.has(peerId)).to.eventually.be.true()
+
+      const peer = await peerStore.get(peerId)
+      expect(peer.addresses.map(({ multiaddr, isCertified }) => ({
+        isCertified,
+        multiaddr: multiaddr.toString()
+      }))).to.deep.equal([{
+        isCertified: true,
+        multiaddr: '/ip4/127.0.0.1/tcp/4567'
+      }])
+    })
+
+    it('ignores older peer records', async () => {
+      const oldSignedPeerRecord = await RecordEnvelope.seal(new PeerRecord({
+        peerId,
+        multiaddrs: [
+          multiaddr('/ip4/127.0.0.1/tcp/1234')
+        ],
+        seqNumber: 1n
+      }), peerId)
+
+      const newSignedPeerRecord = await RecordEnvelope.seal(new PeerRecord({
+        peerId,
+        multiaddrs: [
+          multiaddr('/ip4/127.0.0.1/tcp/4567')
+        ],
+        seqNumber: 2n
+      }), peerId)
+
+      await expect(peerStore.consumePeerRecord(newSignedPeerRecord.marshal())).to.eventually.equal(true)
+      await expect(peerStore.consumePeerRecord(oldSignedPeerRecord.marshal())).to.eventually.equal(false)
+
+      const peer = await peerStore.get(peerId)
+      expect(peer.addresses.map(({ multiaddr, isCertified }) => ({
+        isCertified,
+        multiaddr: multiaddr.toString()
+      }))).to.deep.equal([{
+        isCertified: true,
+        multiaddr: '/ip4/127.0.0.1/tcp/4567'
+      }])
+    })
+
+    it('ignores record for unexpected peer', async () => {
+      const signedPeerRecord = await RecordEnvelope.seal(new PeerRecord({
+        peerId,
+        multiaddrs: [
+          multiaddr('/ip4/127.0.0.1/tcp/4567')
+        ]
+      }), peerId)
+
+      await expect(peerStore.has(peerId)).to.eventually.be.false()
+      await expect(peerStore.consumePeerRecord(signedPeerRecord.marshal(), otherPeerId)).to.eventually.equal(false)
+      await expect(peerStore.has(peerId)).to.eventually.be.false()
+    })
+  })
 })