feat(mesh): protomux v3 wire and persistent friend topic links

Migrate contacts gossip to pearcord-drive/mux-wire (protomux v3).
Add _syncFriendTopicLinks so accepted friends stay subscribed to each
other's contacts topics for live presence and profile cosmetics.

Fix _gossipToPeer race: do not leave persistent friend topics early;
delay ephemeral leave until gossip flush completes. Sync friend topics
before CONTACT_ACCEPT. Expose stats.meshLive and friendTopics count.

Phase 62 / v0.8.27 — contacts mesh polish.

Co-authored-by: Cursor <[email protected]>

index.js

@@ -34,6 +34,7 @@ class PearcordContacts extends EventEmitter {
     this._channels = new Map()
     this._meshJoined = false
     this._ephemeralTopics = new Set()
+    this._friendTopics = new Set()
     this._friendPresence = new Map()
     this._lastSelfPresence = null
   }
@@ -116,6 +117,7 @@ class PearcordContacts extends EventEmitter {
     row.status = CONTACT_STATUS.ACCEPTED
     row.acceptedAt = now()
     await this.store.insert(CONTACTS_COLLECTION, row)
+    await this._syncFriendTopicLinks()
     await this._gossipToPeer(peerUserId, RPC.CONTACT_ACCEPT, {
       ...this._profilePayload(),
       toUserId: peerUserId,
@@ -149,6 +151,7 @@ class PearcordContacts extends EventEmitter {
         toUserId: peerUserId
       }).catch(() => {})
     }
+    await this._syncFriendTopicLinks()
     this.emit('contact', null)
     return row
   }
@@ -240,6 +243,7 @@ class PearcordContacts extends EventEmitter {
       }).catch(() => {})
     }
     await this.store.insert(CONTACTS_COLLECTION, row)
+    if (mutual) await this._syncFriendTopicLinks()
     this.emit('contact-request', row)
     this.emit('contact', row)
     return row
@@ -262,6 +266,7 @@ class PearcordContacts extends EventEmitter {
     row.acceptedAt = now()
     row.peerDisplayName = payload.fromDisplayName || row.peerDisplayName
     await this.store.insert(CONTACTS_COLLECTION, row)
+    await this._syncFriendTopicLinks()
     this.emit('contact', row)
     return row
   }
@@ -298,7 +303,7 @@ class PearcordContacts extends EventEmitter {
     return row
   }
 
-  broadcastPresence (payload) {
+  async broadcastPresence (payload) {
     if (!payload?.userId) return null
     this._lastSelfPresence = {
       userId: payload.userId,
@@ -307,14 +312,14 @@ class PearcordContacts extends EventEmitter {
       at: payload.at || now()
     }
     if (this._meshJoined) {
-      broadcastContactsGossip(this, RPC.PRESENCE_UPDATE, this._lastSelfPresence)
+      await broadcastContactsGossip(this, RPC.PRESENCE_UPDATE, this._lastSelfPresence)
     }
     return this._lastSelfPresence
   }
 
-  broadcastProfileCosmetic (payload) {
+  async broadcastProfileCosmetic (payload) {
     if (!payload?.userId || !this._meshJoined) return null
-    broadcastContactsGossip(this, RPC.PROFILE_COSMETIC_UPDATE, payload)
+    await broadcastContactsGossip(this, RPC.PROFILE_COSMETIC_UPDATE, payload)
     return payload
   }
 
@@ -366,8 +371,25 @@ class PearcordContacts extends EventEmitter {
     return null
   }
 
-  gossipLocal (method, payload) {
-    broadcastContactsGossip(this, method, payload)
+  async gossipLocal (method, payload) {
+    await broadcastContactsGossip(this, method, payload)
+  }
+
+  async _syncFriendTopicLinks () {
+    if (!this.swarm || !this._meshJoined) return
+    const accepted = await this.listAccepted()
+    const want = new Set(accepted.map((r) => contactsTopic(r.peerUserId)))
+    for (const topic of want) {
+      if (this._friendTopics.has(topic)) continue
+      await this.swarm.join(topicToBuffer(topic), { server: false, client: true })
+      this._friendTopics.add(topic)
+    }
+    for (const topic of [...this._friendTopics]) {
+      if (want.has(topic)) continue
+      await this.swarm.leave(topicToBuffer(topic)).catch(() => {})
+      this._friendTopics.delete(topic)
+    }
+    await this._flushSwarm(3500)
   }
 
   async _flushSwarm (ms = Number(process.env.PEARCORD_MESH_FLUSH_MS || 2500)) {
@@ -380,19 +402,21 @@ class PearcordContacts extends EventEmitter {
 
   async _gossipToPeer (peerUserId, method, payload) {
     if (!this.swarm) {
-      this.gossipLocal(method, payload)
+      await this.gossipLocal(method, payload)
       return
     }
     const topic = contactsTopic(peerUserId)
     const buf = topicToBuffer(topic)
-    const wasJoined = this._ephemeralTopics.has(topic)
-    if (!wasJoined) {
+    const persistent = this._friendTopics.has(topic)
+    const wasEphemeral = this._ephemeralTopics.has(topic)
+    if (!persistent && !wasEphemeral) {
       await this.swarm.join(buf, { server: false, client: true })
       this._ephemeralTopics.add(topic)
     }
     await this._flushSwarm()
-    broadcastContactsGossip(this, method, payload)
-    if (!wasJoined) {
+    await broadcastContactsGossip(this, method, payload)
+    if (!persistent && !wasEphemeral) {
+      await new Promise((r) => setTimeout(r, 350))
       await this.swarm.leave(buf).catch(() => {})
       this._ephemeralTopics.delete(topic)
     }
@@ -407,8 +431,8 @@ class PearcordContacts extends EventEmitter {
     this.swarm.on('connection', (conn) => {
       const peerId = b4a.toString(conn.remotePublicKey, 'hex')
       this.peers.set(peerId, conn)
-      const ch = attachContactsMesh(this, conn)
-      this._channels.set(peerId, ch)
+      const wire = attachContactsMesh(this, conn)
+      if (wire) this._channels.set(peerId, wire)
       this.emit('peer', { peerId, type: 'join' })
       conn.on('close', () => {
         this.peers.delete(peerId)
@@ -422,8 +446,9 @@ class PearcordContacts extends EventEmitter {
     })
     await this._flushSwarm(3500)
     this._meshJoined = true
+    await this._syncFriendTopicLinks()
     if (this._lastSelfPresence) {
-      broadcastContactsGossip(this, RPC.PRESENCE_UPDATE, this._lastSelfPresence)
+      await broadcastContactsGossip(this, RPC.PRESENCE_UPDATE, this._lastSelfPresence)
     }
     return this
   }
@@ -432,6 +457,7 @@ class PearcordContacts extends EventEmitter {
     this.peers.clear()
     this._channels.clear()
     this._ephemeralTopics.clear()
+    this._friendTopics.clear()
     this._meshJoined = false
     if (this.swarm) {
       await this.swarm.leave(topicToBuffer(contactsTopic(this.userId))).catch(() => {})
@@ -441,7 +467,12 @@ class PearcordContacts extends EventEmitter {
   }
 
   getStats () {
-    return { peers: this.peers.size }
+    const peers = this.peers.size
+    return {
+      peers,
+      friendTopics: this._friendTopics.size,
+      meshLive: peers > 0
+    }
   }
 }
 

mesh.js

@@ -3,42 +3,41 @@
 const Protomux = require('protomux')
 const b4a = require('b4a')
 const { encodeRpc, decodeRpc } = require('pearcord-shared')
+const { openWireChannel } = require('pearcord-drive/mux-wire')
 
 const CONTACTS_PROTOCOL = 'pearcord-contacts-v1'
 
 function attachContactsMesh (contacts, conn) {
   const mux = Protomux.from(conn)
-  const channel = mux.createChannel({
+  return openWireChannel(mux, {
     protocol: CONTACTS_PROTOCOL,
-    onopen () {
-      channel.open()
-    },
-    onmessage (buf) {
+    ondata (buf) {
       let packet
       try {
         packet = decodeRpc(buf)
         const payload = JSON.parse(b4a.toString(packet.payload))
-        contacts._onGossip(packet.method, payload, conn)
+        contacts._onGossip(packet.method, payload)
       } catch {
         // ignore malformed
       }
     }
   })
-  return channel
 }
 
-function broadcastContactsGossip (contacts, method, payload) {
+async function broadcastContactsGossip (contacts, method, payload) {
   const buf = encodeRpc(method, b4a.from(JSON.stringify(payload)))
-  for (const ch of contacts._channels.values()) {
-    ch.fullyOpened().then((ok) => {
-      if (!ok) return
+  const sessions = [...contacts._channels.values()]
+  await Promise.all(
+    sessions.map(async (session) => {
       try {
-        ch.send(buf)
+        const ok = await session.fullyOpened().catch(() => false)
+        if (!ok) return
+        session.send(buf)
       } catch {
         // peer gone
       }
-    }).catch(() => {})
-  }
+    })
+  )
 }
 
 module.exports = { attachContactsMesh, broadcastContactsGossip, CONTACTS_PROTOCOL }

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pearcord-contacts",
-  "version": "0.1.0",
+  "version": "0.1.1",
   "main": "index.js",
   "type": "commonjs",
   "description": "P2P friend requests and contact list for Pearcord",
@@ -10,6 +10,7 @@
     "hyperswarm": "^4.8.0",
     "protomux": "^3.0.0",
     "pearcord-db": "file:../pearcord-db",
+    "pearcord-drive": "file:../pearcord-drive",
     "pearcord-shared": "file:../pearcord-shared"
   }
 }