| … | … | @@ -19,31 +19,48 @@ |
| 19 | 19 | import b4a from 'b4a' |
| 20 | 20 | import c from 'compact-encoding' |
| 21 | 21 | import Protomux from 'protomux' |
| 22 | import ProtomuxRPC from 'protomux-rpc' |
| 22 | 23 | import { RepoStore } from 'atproto-store/lib/repo-store.js' |
| 23 | 24 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' |
| 24 | | import { decodeCommitFrame } from 'atproto-store/lib/firehose.js' |
| 25 | import { decodeCommitFrame, syncFrame } from 'atproto-store/lib/firehose.js' |
| 25 | 26 | import { getRepoFromCommit, getRecord as rvGetRecord, listRecords as rvListRecords } from 'atproto-store/lib/read-view.js' |
| 26 | 27 | import { readCar, loadCommit, decodeBlock } from 'atproto-repo/lib/repo.js' |
| 27 | 28 | import { verifyCommit, commitCid } from 'atproto-repo/lib/commit.js' |
| 28 | 29 | |
| 29 | 30 | const ANNOUNCE_PROTOCOL = 'atproto/did-announce/v1' |
| 31 | // The peer's application RPC (write/session/blob relay). Opened by DidPeerRouter ONE-per-connection when |
| 32 | // `{ rpc: true }` — the freshest front borrows it. MUST match hyper-proto-peer's serve RPC protocol. |
| 33 | export const RPC_PROTOCOL = 'hyper-proto-peer/rpc/v1' |
| 30 | 34 | const toKey = (k) => (b4a.isBuffer(k) ? k : b4a.from(k, 'hex')) |
| 31 | 35 | |
| 36 | // Reject a promise if it doesn't settle within ms (bounds the router's staleness nudge on a degraded link). |
| 37 | export function withTimeout (p, ms) { |
| 38 | return new Promise((resolve, reject) => { |
| 39 | const t = setTimeout(() => reject(new Error('rpc timeout')), ms) |
| 40 | p.then((v) => { clearTimeout(t); resolve(v) }, (e) => { clearTimeout(t); reject(e) }) |
| 41 | }) |
| 42 | } |
| 43 | |
| 32 | 44 | export class DidPeer { |
| 33 | 45 | // corestore: a Corestore (cores are global by key). did: the DID. verifyKey: its signing pubkey |
| 34 | 46 | // (Uint8Array) — REQUIRED to accept a served head. onHead: (head) => void on every NEW verified |
| 35 | 47 | // head. onError: optional (err) => void. |
| 36 | | constructor (corestore, { did, verifyKey, onHead = null, onError = null } = {}) { |
| 48 | constructor (corestore, { did, verifyKey, handle = null, onHead = null, onError = null } = {}) { |
| 37 | 49 | if (!corestore) throw new Error('DidPeer: corestore required') |
| 38 | 50 | if (!did) throw new Error('DidPeer: did required') |
| 39 | 51 | if (!verifyKey) throw new Error('DidPeer: verifyKey (DID signing pubkey) required — auth is mandatory') |
| 40 | 52 | this.corestore = corestore |
| 41 | 53 | this.did = did |
| 42 | 54 | this.verifyKey = verifyKey |
| 55 | this.handle = handle |
| 43 | 56 | this._onHead = onHead |
| 44 | 57 | this._onError = onError |
| 45 | 58 | this.repo = null |
| 46 | 59 | this.seq = null |
| 60 | // Optional write/session/blob RPC to the peer (peer-signs-over-RPC). Set by DidPeerRouter from the |
| 61 | // per-connection ProtomuxRPC when the router runs with `{ rpc: true }`. Null for a read-only consumer |
| 62 | // (favela) — the keyless read surface below needs no RPC. |
| 63 | this.rpc = null |
| 47 | 64 | this._repoKey = null |
| 48 | 65 | this._firehoseKey = null |
| 49 | 66 | this._channels = new Set() |
| … | … | @@ -206,6 +223,37 @@ export class DidPeer { |
| 206 | 223 | return this.seq.subscribe(cursor, opts) |
| 207 | 224 | } |
| 208 | 225 | |
| 226 | // com.atproto.sync.getRepoStatus -> a fronted peer we replicate + verified is active. |
| 227 | getRepoStatus () { return { did: this.did, active: true, status: 'active' } } |
| 228 | |
| 229 | // com.atproto.repo.describeRepo -> { handle, did, didDoc, collections, handleIsCorrect }. Keyless: |
| 230 | // did/handle come from the announce, collections from the replicated repo, didDoc a best-effort PLC |
| 231 | // fetch (null on failure). doFetch: (url, opts) -> { status, body:Buffer } (the server's wrapper shape). |
| 232 | async describeRepo ({ doFetch } = {}) { |
| 233 | const collections = await this.collections() |
| 234 | let didDoc = null |
| 235 | try { |
| 236 | if (doFetch) { |
| 237 | const r = await doFetch(`https://plc.directory/${this.did}`, { method: 'GET' }) |
| 238 | if (r && r.status === 200 && r.body) didDoc = JSON.parse(r.body.toString()) |
| 239 | } |
| 240 | } catch {} |
| 241 | return { handle: this.handle || null, did: this.did, didDoc, collections, handleIsCorrect: !!this.handle } |
| 242 | } |
| 243 | |
| 244 | // ---- write / session / blob relay (peer-signs-over-RPC) — the PEER holds the key + applies + signs; |
| 245 | // this consumer never holds a key. Requires the router to run with `{ rpc: true }` (this.rpc set). The |
| 246 | // signed result flows back on the peer's firehose (which this DidPeer already replicates + reconciles). |
| 247 | applyWrite (op) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('apply-write', op) } |
| 248 | applyWrites (op) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('apply-writes', op) } |
| 249 | createSession ({ identifier, password } = {}) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('create-session', { identifier, password }) } |
| 250 | refreshSession (refreshJwt) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('refresh-session', { refreshJwt }) } |
| 251 | getServiceAuth ({ aud, lxm, token } = {}) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('get-service-auth', { aud, lxm, token }) } |
| 252 | getPreferences ({ token } = {}) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('get-preferences', { token }) } |
| 253 | putPreferences ({ preferences, token } = {}) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('put-preferences', { preferences, token }) } |
| 254 | getBlob (cid) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('get-blob', { cid }) } |
| 255 | uploadBlob ({ bytes, mimeType, token } = {}) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('upload-blob', { bytes, mimeType, token }) } |
| 256 | |
| 209 | 257 | // Decode a #commit frame BODY (from firehose.decodeFrame) into normalized ops with the changed record |
| 210 | 258 | // INLINE — read from the frame's OWN CAR (body.blocks) keyed by op.cid, so it is race-free (does NOT |
| 211 | 259 | // depend on the repo core having caught up). Returns [{ action, collection, rkey, record? }]; delete |
| … | … | @@ -263,41 +311,119 @@ export class DidPeer { |
| 263 | 311 | } |
| 264 | 312 | } |
| 265 | 313 | |
| 266 | | // Multi-DID reader over ONE shared corestore + swarm. Fixes the per-DidPeer attach() collision: each |
| 267 | | // DidPeer.attach opens its OWN ANNOUNCE_PROTOCOL channel per connection, but Protomux allows only ONE |
| 268 | | // channel per (protocol, id) on a mux — so when several DidPeers share a swarm, only the FIRST-added |
| 269 | | // peer binds the channel and every other DID silently never learns its cores (its announce is received |
| 270 | | // by the first peer, which drops it as a did mismatch). The router opens ONE announce channel per |
| 271 | | // connection and DISPATCHES each announce to the matching DidPeer by did. Mirrors atproto-bridge's |
| 272 | | // PeerRouter / keet's one-channel-per-connection model. |
| 314 | // Multi-DID, co-owned-aware router over ONE shared corestore + swarm — the CANONICAL consumer primitive |
| 315 | // (unifies atproto-bridge's PeerRouter/PeerFront + favela's earlier single-front DidPeerRouter, per |
| 316 | // EXTRACTION-PLAN Phase 1). Fixes the per-DidPeer attach() channel collision (Protomux allows one channel |
| 317 | // per (protocol,id) per mux) by opening ONE announce channel per connection and dispatching by DID. |
| 273 | 318 | // |
| 274 | | // The caller owns the swarm (and joins discoveryTopic(did) per DID); the router owns the corestore |
| 275 | | // replication + the shared announce channel. Wire it: swarm.on('connection', c => router.onconnection(c)). |
| 319 | // Beyond that, it absorbs PeerRouter's co-owned handling: a DID can have MANY self-custody holders (wj62 = |
| 320 | // notary + co-owner; each announces the SAME did with its OWN membrane core keys — distinct keys, identical |
| 321 | // deterministic content, differing only in replication lag). We track a SET of fronts per DID (one per |
| 322 | // firehoseKey) and serve reads from the FRESHEST by commit REV (not frame count — a lagging co-owner can |
| 323 | // out-FRAME a fresher one; the wj62 prod symptom). Optional layers, all off by default (favela's read-only |
| 324 | // posture is `new DidPeerRouter(store)`): |
| 325 | // swarm — router OWNS it: auto-wires onconnection + a pull-only refresh/failover timer. Omit to drive |
| 326 | // onconnection() yourself and join topics yourself (favela / tests). |
| 327 | // join — (did)=>void topic-join fn (bridge passes `d=>swarm.join(discoveryTopic(d),{...})`); keeps |
| 328 | // atproto-identity OUT of this module's runtime deps. |
| 329 | // fronted — DIDs to allow-list immediately (bridge). resolveVerifyKey — async (did)=>pubkey fallback. |
| 330 | // relay — FleetRelay: fold the FRESHEST co-owner's firehose per DID (single source) + failover + #sync. |
| 331 | // rpc — open ONE protomux-rpc per connection (the write/session/blob relay); the front borrows it. |
| 332 | // handles/onHandles — persist handle->did resolution across restarts. |
| 276 | 333 | export class DidPeerRouter { |
| 277 | | constructor (corestore) { |
| 334 | constructor (corestore, { |
| 335 | swarm = null, fronted = [], relay = null, resolveVerifyKey = null, rpc = false, join = null, |
| 336 | refreshMs = 10000, refreshTimeoutMs = 5000, handles = {}, onHandles = null |
| 337 | } = {}) { |
| 278 | 338 | this.corestore = corestore |
| 279 | | this._peers = new Map() // did -> DidPeer |
| 339 | this.swarm = swarm |
| 340 | this.relay = relay |
| 341 | this._resolveVerifyKey = resolveVerifyKey |
| 342 | this._useRpc = !!rpc |
| 343 | this._join = join |
| 344 | this._refreshMs = refreshMs |
| 345 | this._refreshTimeoutMs = refreshTimeoutMs |
| 346 | |
| 347 | this._peers = new Map() // did -> primary DidPeer (backward-compat add()/get()) |
| 348 | this._fronts = new Map() // did -> Map<fhKeyHex, DidPeer> (co-owned multi-front) |
| 349 | this._byDid = new Map() // did -> Map<fhKeyHex, { conn, rpc, coords }> (identified connections) |
| 350 | this._opts = new Map() // did -> { verifyKey, handle, onHead, onError } from add() |
| 351 | this._verifyKeys = new Map() // did -> Promise<pubkey> |
| 352 | this._fronted = new Set(fronted) |
| 353 | this._opening = new Map() // "did|fhKeyHex" -> in-flight open promise |
| 354 | this._waiters = new Map() // did -> Set<fn> woken when ANY co-owner of that did lands |
| 355 | this._lastAppend = new Map() // did -> ts of last firehose 'append' (event-driven liveness) |
| 356 | this._relaySourceKey = new Map() // did -> firehoseKeyHex currently folded into the relay |
| 357 | this._handleToDid = new Map(Object.entries(handles || {})) |
| 358 | this._onHandles = onHandles |
| 280 | 359 | this._channels = new Set() |
| 360 | |
| 361 | if (this.swarm) { |
| 362 | this._onconnbound = (conn) => this.onconnection(conn) |
| 363 | this.swarm.on('connection', this._onconnbound) |
| 364 | } |
| 365 | for (const did of this._fronted) this._joinDid(did) |
| 366 | if (this._refreshMs > 0 && (this.swarm || this.relay)) { |
| 367 | this._refreshTimer = setInterval(() => { this._refreshFronts().catch(() => {}) }, this._refreshMs) |
| 368 | if (this._refreshTimer && this._refreshTimer.unref) this._refreshTimer.unref() |
| 369 | } |
| 281 | 370 | } |
| 282 | 371 | |
| 283 | 372 | get (did) { return this._peers.get(did) || null } |
| 284 | 373 | peers () { return [...this._peers.values()] } |
| 374 | isFronted (did) { return !!did && this._fronted.has(did) } |
| 375 | list () { return [...this._fronted] } |
| 376 | resolveHandle (handle) { return this._handleToDid.get(handle) || null } |
| 377 | |
| 378 | _isCared (did) { return this._fronted.has(did) || this._peers.has(did) || this._opts.has(did) } |
| 379 | _joinDid (did) { if (this._join) { try { this._join(did) } catch {} } } |
| 380 | _wake (did) { const ws = this._waiters.get(did); if (ws) { this._waiters.delete(did); for (const fn of ws) fn() } } |
| 381 | |
| 382 | _setHandle (handle, did) { |
| 383 | if (!handle || !did || this._handleToDid.get(handle) === did) return |
| 384 | this._handleToDid.set(handle, did) |
| 385 | if (this._onHandles) { try { this._onHandles(Object.fromEntries(this._handleToDid)) } catch {} } |
| 386 | } |
| 387 | |
| 388 | // Cached DID->verifyKey. Per-DID add({verifyKey}) wins; else the injected resolver (PLC). |
| 389 | _verifyKey (did) { |
| 390 | const o = this._opts.get(did) || {} |
| 391 | if (o.verifyKey) return Promise.resolve(o.verifyKey) |
| 392 | if (!this._verifyKeys.has(did)) { |
| 393 | if (!this._resolveVerifyKey) return Promise.reject(new Error(`DidPeerRouter: no verifyKey for ${did} (pass add(did,{verifyKey}) or {resolveVerifyKey})`)) |
| 394 | this._verifyKeys.set(did, Promise.resolve(this._resolveVerifyKey(did))) |
| 395 | } |
| 396 | return this._verifyKeys.get(did) |
| 397 | } |
| 285 | 398 | |
| 286 | | // Register (or return) a DidPeer for a did. Does NOT join the swarm — the caller joins the topic. |
| 399 | // Register a DID. With a verifyKey (favela) it eagerly creates + returns the primary DidPeer (its cores |
| 400 | // load from the announce); without one (bridge allow-set) it just allow-lists + returns `this` (chainable, |
| 401 | // verifyKey resolved lazily at front time). Idempotent. |
| 287 | 402 | add (did, opts = {}) { |
| 288 | | let p = this._peers.get(did) |
| 289 | | if (p) return p |
| 290 | | p = new DidPeer(this.corestore, { did, ...opts }) |
| 403 | this._fronted.add(did) |
| 404 | if (opts.verifyKey || opts.onHead || opts.onError || opts.handle) { |
| 405 | this._opts.set(did, { ...(this._opts.get(did) || {}), ...opts }) |
| 406 | } |
| 407 | this._joinDid(did) |
| 408 | const existing = this._peers.get(did) |
| 409 | if (existing) return existing |
| 410 | const o = this._opts.get(did) || {} |
| 411 | if (!o.verifyKey) return this // bridge allow-set add (verifyKey lazy); return value unused/chainable |
| 412 | const p = new DidPeer(this.corestore, { did, verifyKey: o.verifyKey, handle: o.handle || null, onHead: o.onHead || null, onError: o.onError || null }) |
| 291 | 413 | this._peers.set(did, p) |
| 292 | 414 | for (const chan of this._channels) p.announceVia(chan._announce) // (re-add) announce if already held |
| 293 | 415 | return p |
| 294 | 416 | } |
| 295 | 417 | |
| 296 | | // Wire ONE swarm connection: replicate the corestore ONCE, open ONE announce channel, and route each |
| 297 | | // received announce to the matching DidPeer by did. |
| 418 | // Wire ONE swarm connection: replicate the corestore ONCE, open ONE announce channel (+ one write RPC |
| 419 | // when `{ rpc: true }`), and route each received announce by DID -> (did, firehoseKey) tracking. |
| 298 | 420 | onconnection (conn) { |
| 299 | 421 | try { this.corestore.replicate(conn) } catch {} |
| 300 | 422 | const mux = Protomux.from(conn) |
| 423 | let rpc = null |
| 424 | if (this._useRpc) { |
| 425 | try { rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json }); rpc.on('error', () => {}) } catch {} |
| 426 | } |
| 301 | 427 | const chan = mux.createChannel({ |
| 302 | 428 | protocol: ANNOUNCE_PROTOCOL, |
| 303 | 429 | onopen: () => this._announceHeld(chan), |
| … | … | @@ -306,26 +432,221 @@ export class DidPeerRouter { |
| 306 | 432 | if (!chan) return // already open on this mux |
| 307 | 433 | chan._announce = chan.addMessage({ |
| 308 | 434 | encoding: c.json, |
| 309 | | onmessage: (a) => { |
| 310 | | const p = a && a.did ? this._peers.get(a.did) : null |
| 311 | | if (p) p.onAnnounce(a).catch(() => {}) |
| 312 | | } |
| 435 | onmessage: (a) => { this._onAnnounce(a, rpc, conn).catch(() => {}) } |
| 313 | 436 | }) |
| 314 | 437 | chan.open() |
| 315 | 438 | this._channels.add(chan) |
| 316 | 439 | this._announceHeld(chan) |
| 440 | conn.on('close', () => this._onConnClose(conn)) |
| 441 | } |
| 442 | |
| 443 | // Learn a (did, firehoseKey) co-owner from a PUSHED announce; open its front (best-effort). A DIFFERENT |
| 444 | // firehoseKey under the same DID is a DISTINCT co-owner (tracked, never evicting the other). |
| 445 | async _onAnnounce (a, rpc, conn) { |
| 446 | if (!a || !a.did || !a.repoKey || !a.firehoseKey) return |
| 447 | const did = a.did |
| 448 | const fhKeyHex = a.firehoseKey |
| 449 | if (a.handle) this._setHandle(a.handle, did) |
| 450 | if (!this._isCared(did)) return |
| 451 | if (!this._byDid.has(did)) this._byDid.set(did, new Map()) |
| 452 | this._byDid.get(did).set(fhKeyHex, { conn, rpc, coords: a }) |
| 453 | const existing = this._fronts.get(did) && this._fronts.get(did).get(fhKeyHex) |
| 454 | if (existing) { // refresh write rpc + handle; nudge fetch (cores re-replicate via corestore regardless) |
| 455 | if (this._useRpc && rpc) existing.rpc = rpc |
| 456 | if (a.handle) existing.handle = a.handle |
| 457 | if (existing.seq) existing.seq.core.update().catch(() => {}) |
| 458 | this._wake(did) |
| 459 | return |
| 460 | } |
| 461 | this._wake(did) |
| 462 | this._ensureFront(did, fhKeyHex).catch(() => {}) |
| 463 | } |
| 464 | |
| 465 | // Open (once, dedup-guarded) the front for ONE co-owner (did, firehoseKey). Reuses the primary DidPeer |
| 466 | // (from add()) for its FIRST co-owner; a co-owner with a different firehoseKey gets its own DidPeer. |
| 467 | async _ensureFront (did, fhKeyHex) { |
| 468 | const set = this._fronts.get(did) |
| 469 | if (set && set.has(fhKeyHex)) return set.get(fhKeyHex) |
| 470 | const openKey = did + '|' + fhKeyHex |
| 471 | const inflight = this._opening.get(openKey) |
| 472 | if (inflight) return inflight |
| 473 | const entry = this._byDid.get(did) && this._byDid.get(did).get(fhKeyHex) |
| 474 | if (!entry) return null |
| 475 | const p = (async () => { |
| 476 | const verifyKey = await this._verifyKey(did) |
| 477 | const o = this._opts.get(did) || {} |
| 478 | let peer = this._peers.get(did) |
| 479 | if (peer && peer.holds) { |
| 480 | const boundFh = peer._firehoseKey ? b4a.toString(peer._firehoseKey, 'hex') : null |
| 481 | if (boundFh !== fhKeyHex) peer = null // primary already bound to a different co-owner -> new front |
| 482 | } |
| 483 | if (!peer) peer = new DidPeer(this.corestore, { did, verifyKey, handle: entry.coords.handle || o.handle || null, onHead: o.onHead || null, onError: o.onError || null }) |
| 484 | if (entry.coords.handle) peer.handle = entry.coords.handle |
| 485 | if (this._useRpc && entry.rpc) peer.rpc = entry.rpc |
| 486 | await peer.provide({ repoKey: entry.coords.repoKey, firehoseKey: entry.coords.firehoseKey }) |
| 487 | if (!this._fronts.has(did)) this._fronts.set(did, new Map()) |
| 488 | this._fronts.get(did).set(fhKeyHex, peer) |
| 489 | if (!this._peers.has(did)) this._peers.set(did, peer) // first front becomes the primary (get()) |
| 490 | if (peer.seq && peer.seq.core && !peer._appendWired) { |
| 491 | peer._appendWired = true |
| 492 | this._lastAppend.set(did, Date.now()) |
| 493 | peer.seq.core.on('append', () => { this._lastAppend.set(did, Date.now()) }) |
| 494 | } |
| 495 | if (peer.handle) this._setHandle(peer.handle, did) |
| 496 | this._wake(did) |
| 497 | this._pickRelaySource(did).catch(() => {}) |
| 498 | return peer |
| 499 | })().catch(() => null) |
| 500 | this._opening.set(openKey, p) |
| 501 | try { return await p } finally { this._opening.delete(openKey) } |
| 502 | } |
| 503 | |
| 504 | // The freshest VERIFIED front for a DID = the co-owner whose head has the highest REV (a head is only set |
| 505 | // after its commit verifies against the DID key). Prefer a CURRENTLY-announced (live) front; fall back to |
| 506 | // any verified front if none are announced (all co-owners momentarily disconnected). Revs are TID strings |
| 507 | // (lexicographic == chronological). Never regresses to a lagging co-owner once a fresher one is open. |
| 508 | _freshestFront (did) { |
| 509 | const set = this._fronts.get(did) |
| 510 | if (!set || !set.size) return null |
| 511 | const live = this._byDid.get(did) || new Map() |
| 512 | const scored = [] |
| 513 | for (const [fhKeyHex, f] of set) { |
| 514 | if (!f || !f.head || !f.head.rev) continue // only verified fronts |
| 515 | scored.push({ f, rev: f.head.rev, live: live.has(fhKeyHex) }) |
| 516 | } |
| 517 | if (!scored.length) return null |
| 518 | const pickMaxRev = (arr) => arr.reduce((b, x) => (!b || x.rev > b.rev ? x : b), null) |
| 519 | const liveOnes = scored.filter((x) => x.live) |
| 520 | const chosen = pickMaxRev(liveOnes) || pickMaxRev(scored) |
| 521 | return chosen ? chosen.f : null |
| 522 | } |
| 523 | |
| 524 | // Get the FRESHEST verified front for a fronted DID: ensure fronts for every identified co-owner, then |
| 525 | // return the highest-rev one. This is what every read (getRepo/listRecords/…) resolves against — a |
| 526 | // lagging co-owner never freezes the served head. |
| 527 | async front (did, { timeout } = {}) { |
| 528 | if (!this._isCared(did)) throw new Error(`DidPeerRouter: ${did} is not a fronted DID`) |
| 529 | this._joinDid(did) |
| 530 | if (!this._byDid.has(did) || this._byDid.get(did).size === 0) await this._waitForDid(did, timeout || 20000) |
| 531 | const entries = this._byDid.get(did) || new Map() |
| 532 | await Promise.all([...entries.keys()].map((fh) => this._ensureFront(did, fh).catch(() => null))) |
| 533 | const best = this._freshestFront(did) |
| 534 | if (!best) throw new Error(`DidPeerRouter: no frontable co-owner for ${did}`) |
| 535 | return best |
| 536 | } |
| 537 | |
| 538 | // Resolve that AT LEAST ONE co-owner of a DID is identified — immediately if any is connected, else wait. |
| 539 | _waitForDid (did, timeout) { |
| 540 | const e = this._byDid.get(did) |
| 541 | if (e && e.size > 0) return Promise.resolve(e) |
| 542 | return new Promise((resolve, reject) => { |
| 543 | const fn = () => { |
| 544 | clearTimeout(timer) |
| 545 | const e2 = this._byDid.get(did) |
| 546 | e2 && e2.size > 0 ? resolve(e2) : reject(new Error(`DidPeerRouter: ${did} identified but entry missing`)) |
| 547 | } |
| 548 | const timer = setTimeout(() => { |
| 549 | const ws = this._waiters.get(did); if (ws) ws.delete(fn) |
| 550 | reject(new Error(`DidPeerRouter: timeout finding ${did}`)) |
| 551 | }, timeout) |
| 552 | if (!this._waiters.has(did)) this._waiters.set(did, new Set()) |
| 553 | this._waiters.get(did).add(fn) |
| 554 | }) |
| 555 | } |
| 556 | |
| 557 | // A peer disconnected: drop its (did, firehoseKey) identified-connection entries so _onAnnounce re-runs |
| 558 | // on reconnect. Do NOT close the cached front — its cores keep replicating over any remaining/next conn. |
| 559 | _onConnClose (conn) { |
| 560 | for (const [did, entries] of this._byDid) { |
| 561 | for (const [fhKeyHex, entry] of entries) { if (entry.conn === conn) entries.delete(fhKeyHex) } |
| 562 | if (entries.size === 0) this._byDid.delete(did) |
| 563 | } |
| 564 | } |
| 565 | |
| 566 | // Fold the FRESHEST co-owner's firehose into the aggregate relay (single source per DID — two co-owners |
| 567 | // are the SAME frames under different sourceIds, so folding both duplicates every #commit). On a source |
| 568 | // change (fresher co-owner won, or the folded one stalled) swap removeSource->addSource + emit #sync. |
| 569 | async _pickRelaySource (did) { |
| 570 | if (!this.relay) return |
| 571 | const best = this._freshestFront(did) |
| 572 | if (!best || !best.seq || !best.seq.core) return |
| 573 | const key = b4a.toString(best.seq.core.key, 'hex') |
| 574 | if (this._relaySourceKey.get(did) === key) return |
| 575 | const old = this._relaySourceKey.get(did) |
| 576 | this._relaySourceKey.set(did, key) |
| 577 | ;(async () => { |
| 578 | if (old) { try { this.relay.removeSource(old) } catch {} } |
| 579 | try { await this.relay.addSource(key, best.seq.core) } catch {} |
| 580 | try { await this._emitSync(did, best) } catch {} |
| 581 | })().catch(() => {}) |
| 582 | } |
| 583 | |
| 584 | async _emitSync (did, front) { |
| 585 | if (!this.relay) return { did, emitted: false, reason: 'no relay' } |
| 586 | const signed = front.head && front.head.signed |
| 587 | if (!signed) return { did, emitted: false, reason: 'no commit' } |
| 588 | const time = new Date().toISOString() |
| 589 | const frameBytes = await syncFrame({ seq: 0, did, signedCommit: signed, time }) |
| 590 | const emitted = await this.relay.emit(`sync:${did}:${signed.rev}`, frameBytes) |
| 591 | return { did, rev: signed.rev, emitted } |
| 592 | } |
| 593 | |
| 594 | // Public control-plane trigger: force a #sync re-baseline for a fronted DID (from its freshest front). |
| 595 | async emitSync (did) { const f = await this.front(did); return this._emitSync(did, f) } |
| 596 | |
| 597 | // Pull-only staleness ceiling + relay failover (owns-swarm/relay mode). NEVER destroys a connection — |
| 598 | // cores live-follow via the standing download; this just nudges a front that saw no 'append' for a window |
| 599 | // and re-picks the relay source if the folded co-owner stalled while another advanced. |
| 600 | async _refreshFronts () { |
| 601 | for (const [did, set] of this._fronts) { |
| 602 | for (const [, f] of set) { |
| 603 | if (!f || !f.seq || !f.seq.core) continue |
| 604 | if (Date.now() - (this._lastAppend.get(did) || 0) < this._refreshMs) continue |
| 605 | try { await withTimeout(f.seq.core.update({ wait: true }), this._refreshTimeoutMs) } catch {} |
| 606 | } |
| 607 | this._pickRelaySource(did).catch(() => {}) |
| 608 | } |
| 609 | } |
| 610 | |
| 611 | diagnose () { |
| 612 | const fronts = [] |
| 613 | for (const [did, set] of this._fronts) { |
| 614 | for (const [fhKeyHex, f] of set) { |
| 615 | fronts.push({ |
| 616 | did, |
| 617 | handle: f.handle || null, |
| 618 | repoKey: f._repoKey ? b4a.toString(f._repoKey, 'hex') : null, |
| 619 | firehoseKey: fhKeyHex, |
| 620 | firehoseLen: (f.seq && f.seq.core) ? f.seq.core.length : null, |
| 621 | rev: f.head ? f.head.rev : null, |
| 622 | rpcConnected: !!(f.rpc && !f.rpc.closed), |
| 623 | relaySource: this._relaySourceKey.get(did) === fhKeyHex, |
| 624 | live: (this._byDid.get(did) || new Map()).has(fhKeyHex) |
| 625 | }) |
| 626 | } |
| 627 | } |
| 628 | return { |
| 629 | fronted: [...this._fronted], |
| 630 | cachedFronts: fronts.length, |
| 631 | fronts, |
| 632 | relay: this.relay ? { sources: this.relay.sourceCount ? this.relay.sourceCount() : null } : null |
| 633 | } |
| 317 | 634 | } |
| 318 | 635 | |
| 319 | | // Announce every held DID on a channel (symmetric seeding: tell the peer which cores we can serve). |
| 636 | // Announce every HELD DID on a channel (symmetric seeding: tell the peer which cores we can serve). |
| 320 | 637 | _announceHeld (chan) { |
| 321 | | for (const p of this._peers.values()) p.announceVia(chan._announce) |
| 638 | for (const p of this._peers.values()) { if (p && p.holds) p.announceVia(chan._announce) } |
| 322 | 639 | } |
| 323 | 640 | |
| 324 | 641 | async close () { |
| 642 | if (this._refreshTimer) { try { clearInterval(this._refreshTimer) } catch {} this._refreshTimer = null } |
| 643 | if (this.swarm && this._onconnbound) { try { this.swarm.off('connection', this._onconnbound) } catch {} } |
| 325 | 644 | for (const chan of this._channels) { try { chan.close() } catch {} } |
| 326 | 645 | this._channels.clear() |
| 327 | | for (const p of this._peers.values()) { try { await p.close() } catch {} } |
| 328 | | this._peers.clear() |
| 646 | const closed = new Set() |
| 647 | for (const set of this._fronts.values()) { for (const f of set.values()) { if (closed.has(f)) continue; closed.add(f); try { await f.close() } catch {} } } |
| 648 | for (const p of this._peers.values()) { if (closed.has(p)) continue; closed.add(p); try { await p.close() } catch {} } |
| 649 | this._fronts.clear(); this._peers.clear(); this._byDid.clear() |
| 329 | 650 | } |
| 330 | 651 | } |
| 331 | 652 | |