| … | … | @@ -89,13 +89,31 @@ export class DidPeer { |
| 89 | 89 | if (!chan) return // already open on this mux |
| 90 | 90 | chan._announce = chan.addMessage({ |
| 91 | 91 | encoding: c.json, |
| 92 | | onmessage: (a) => this._onannounce(a).catch((e) => this._fail(e)) |
| 92 | onmessage: (a) => this.onAnnounce(a).catch((e) => this._fail(e)) |
| 93 | 93 | }) |
| 94 | 94 | chan.open() |
| 95 | 95 | this._channels.add(chan) |
| 96 | 96 | if (this.repo) this._announceTo(chan) |
| 97 | 97 | } |
| 98 | 98 | |
| 99 | // Does this peer currently hold (and can serve/seed) the DID's cores? |
| 100 | get holds () { return !!this.repo } |
| 101 | |
| 102 | // Send this peer's announce over a protomux message (used by DidPeerRouter's shared per-connection |
| 103 | // channel, where a single channel is multiplexed across many DIDs). |
| 104 | announceVia (message) { |
| 105 | if (!this.repo || !message) return |
| 106 | try { |
| 107 | message.send({ |
| 108 | did: this.did, |
| 109 | repoKey: b4a.toString(this._repoKey, 'hex'), |
| 110 | firehoseKey: b4a.toString(this._firehoseKey, 'hex'), |
| 111 | repoLength: this.repo.core.length, |
| 112 | firehoseLength: this.seq.core.length |
| 113 | }) |
| 114 | } catch (e) { this._fail(e) } |
| 115 | } |
| 116 | |
| 99 | 117 | _announceTo (chan) { |
| 100 | 118 | if (!this.repo || !chan._announce) return |
| 101 | 119 | try { |
| … | … | @@ -111,7 +129,7 @@ export class DidPeer { |
| 111 | 129 | |
| 112 | 130 | _announceAll () { for (const chan of this._channels) this._announceTo(chan) } |
| 113 | 131 | |
| 114 | | async _onannounce (a) { |
| 132 | async onAnnounce (a) { |
| 115 | 133 | if (!a || a.did !== this.did || !a.repoKey || !a.firehoseKey) return |
| 116 | 134 | if (!this.repo) { |
| 117 | 135 | await this._openCores(a.repoKey, a.firehoseKey) |
| … | … | @@ -240,4 +258,70 @@ export class DidPeer { |
| 240 | 258 | } |
| 241 | 259 | } |
| 242 | 260 | |
| 261 | // Multi-DID reader over ONE shared corestore + swarm. Fixes the per-DidPeer attach() collision: each |
| 262 | // DidPeer.attach opens its OWN ANNOUNCE_PROTOCOL channel per connection, but Protomux allows only ONE |
| 263 | // channel per (protocol, id) on a mux — so when several DidPeers share a swarm, only the FIRST-added |
| 264 | // peer binds the channel and every other DID silently never learns its cores (its announce is received |
| 265 | // by the first peer, which drops it as a did mismatch). The router opens ONE announce channel per |
| 266 | // connection and DISPATCHES each announce to the matching DidPeer by did. Mirrors atproto-bridge's |
| 267 | // PeerRouter / keet's one-channel-per-connection model. |
| 268 | // |
| 269 | // The caller owns the swarm (and joins discoveryTopic(did) per DID); the router owns the corestore |
| 270 | // replication + the shared announce channel. Wire it: swarm.on('connection', c => router.onconnection(c)). |
| 271 | export class DidPeerRouter { |
| 272 | constructor (corestore) { |
| 273 | this.corestore = corestore |
| 274 | this._peers = new Map() // did -> DidPeer |
| 275 | this._channels = new Set() |
| 276 | } |
| 277 | |
| 278 | get (did) { return this._peers.get(did) || null } |
| 279 | peers () { return [...this._peers.values()] } |
| 280 | |
| 281 | // Register (or return) a DidPeer for a did. Does NOT join the swarm — the caller joins the topic. |
| 282 | add (did, opts = {}) { |
| 283 | let p = this._peers.get(did) |
| 284 | if (p) return p |
| 285 | p = new DidPeer(this.corestore, { did, ...opts }) |
| 286 | this._peers.set(did, p) |
| 287 | for (const chan of this._channels) p.announceVia(chan._announce) // (re-add) announce if already held |
| 288 | return p |
| 289 | } |
| 290 | |
| 291 | // Wire ONE swarm connection: replicate the corestore ONCE, open ONE announce channel, and route each |
| 292 | // received announce to the matching DidPeer by did. |
| 293 | onconnection (conn) { |
| 294 | try { this.corestore.replicate(conn) } catch {} |
| 295 | const mux = Protomux.from(conn) |
| 296 | const chan = mux.createChannel({ |
| 297 | protocol: ANNOUNCE_PROTOCOL, |
| 298 | onopen: () => this._announceHeld(chan), |
| 299 | onclose: () => this._channels.delete(chan) |
| 300 | }) |
| 301 | if (!chan) return // already open on this mux |
| 302 | chan._announce = chan.addMessage({ |
| 303 | encoding: c.json, |
| 304 | onmessage: (a) => { |
| 305 | const p = a && a.did ? this._peers.get(a.did) : null |
| 306 | if (p) p.onAnnounce(a).catch(() => {}) |
| 307 | } |
| 308 | }) |
| 309 | chan.open() |
| 310 | this._channels.add(chan) |
| 311 | this._announceHeld(chan) |
| 312 | } |
| 313 | |
| 314 | // Announce every held DID on a channel (symmetric seeding: tell the peer which cores we can serve). |
| 315 | _announceHeld (chan) { |
| 316 | for (const p of this._peers.values()) p.announceVia(chan._announce) |
| 317 | } |
| 318 | |
| 319 | async close () { |
| 320 | for (const chan of this._channels) { try { chan.close() } catch {} } |
| 321 | this._channels.clear() |
| 322 | for (const p of this._peers.values()) { try { await p.close() } catch {} } |
| 323 | this._peers.clear() |
| 324 | } |
| 325 | } |
| 326 | |
| 243 | 327 | export { ANNOUNCE_PROTOCOL } |