| … | … | @@ -35,26 +35,63 @@ export function attachPresence (conn, { selfKey, presence, id = null, getHead = |
| 35 | 35 | const self = String(selfKey) |
| 36 | 36 | let peerKey = null |
| 37 | 37 | let done = false |
| 38 | let chan = null |
| 39 | let msg = null |
| 38 | 40 | const clear = () => { if (peerKey) presence.markAbsent(peerKey) } |
| 41 | const announce = (headRev) => { |
| 42 | if (!msg) return |
| 43 | try { msg.send({ key: self, headRev: headRev !== undefined ? (headRev || null) : (getHead ? getHead() : null) }) } catch {} |
| 44 | } |
| 39 | 45 | try { |
| 40 | | const chan = Protomux.from(conn).createChannel({ protocol: PRESENCE_PROTOCOL, id }) |
| 41 | | if (!chan) return noop // duplicate (protocol,id) on this mux → harmless no-op |
| 42 | | const msg = chan.addMessage({ |
| 43 | | encoding: c.json, |
| 44 | | onmessage: (data) => { |
| 45 | | if (!data || !data.key) return |
| 46 | | peerKey = String(data.key) |
| 47 | | if (onReceive) { try { onReceive(peerKey) } catch {} } |
| 48 | | presence.markPresent(peerKey) // presence (fires onChange on connect) |
| 49 | | if (data.headRev !== undefined) presence.noteHead(peerKey, data.headRev || null) // currency (fires onChange on a head change) |
| 50 | | } |
| 51 | | }) |
| 52 | | chan.open() |
| 53 | | const announce = (headRev) => { try { msg.send({ key: self, headRev: headRev !== undefined ? (headRev || null) : (getHead ? getHead() : null) }) } catch {} } |
| 54 | | announce() // initial announce with our current head |
| 46 | const mux = Protomux.from(conn) |
| 47 | // ⛔ A LATE ATTACH MUST STILL PAIR (2026-10-14, loser-club). Protomux REJECTS a remote open that finds no local |
| 48 | // channel and no pair() handler (_requestSession → _rejectSession), and the rejected side's channel closes. |
| 49 | // A DID that boots after the connection is up (a resync side store) attached after its peer had already |
| 50 | // opened: the peer's open was rejected, then ours hit a peer with no channel and no handler and was rejected |
| 51 | // too — both attached, both received nothing, both elected themselves. So the channel is (re)opened on |
| 52 | // demand: once now, and again from pair() whenever the peer opens while we have no live channel. |
| 53 | const openChannel = () => { |
| 54 | if (done) return false |
| 55 | if (chan) return true |
| 56 | const ch = mux.createChannel({ |
| 57 | protocol: PRESENCE_PROTOCOL, |
| 58 | id, |
| 59 | onclose: () => { |
| 60 | if (chan !== ch) return |
| 61 | chan = null |
| 62 | msg = null |
| 63 | // A channel that closes with the connection still up (rejected, or the peer detached) loses the peer. |
| 64 | if (peerKey) { presence.markAbsent(peerKey); peerKey = null } |
| 65 | } |
| 66 | }) |
| 67 | if (!ch) return false // duplicate (protocol,id) on this mux → another attach owns it |
| 68 | const m = ch.addMessage({ |
| 69 | encoding: c.json, |
| 70 | onmessage: (data) => { |
| 71 | if (!data || !data.key) return |
| 72 | peerKey = String(data.key) |
| 73 | if (onReceive) { try { onReceive(peerKey) } catch {} } |
| 74 | presence.markPresent(peerKey) // presence (fires onChange on connect) |
| 75 | if (data.headRev !== undefined) presence.noteHead(peerKey, data.headRev || null) // currency (fires onChange on a head change) |
| 76 | } |
| 77 | }) |
| 78 | chan = ch |
| 79 | msg = m |
| 80 | ch.open() |
| 81 | announce() // initial announce with our current head |
| 82 | return true |
| 83 | } |
| 84 | if (!openChannel()) return noop // duplicate → harmless no-op (the first attach owns pairing too) |
| 85 | mux.pair({ protocol: PRESENCE_PROTOCOL, id }, () => { openChannel() }) |
| 55 | 86 | const onclose = () => { if (done) return; done = true; clear() } |
| 56 | 87 | conn.on('close', onclose) |
| 57 | | const detach = () => { if (done) return; done = true; try { conn.off('close', onclose) } catch {}; clear() } |
| 88 | const detach = () => { |
| 89 | if (done) return |
| 90 | done = true |
| 91 | try { mux.unpair({ protocol: PRESENCE_PROTOCOL, id }) } catch {} |
| 92 | try { conn.off('close', onclose) } catch {} |
| 93 | clear() |
| 94 | } |
| 58 | 95 | detach.send = (headRev) => announce(headRev) // re-announce our head (currency) when it changes |
| 59 | 96 | detach.attached = true |
| 60 | 97 | return detach |