// PeerRouter: per-DID routing to fronted hyper-proto-peers — isFronted gate, discover-on-first-use,
// caching, and rejection of non-fronted DIDs. Uses a real (testnet) discoverable peer + an
// injected verifyKey resolver (the PLC fetch is exercised separately at migration time).
import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import b4a from 'b4a'
import c from 'compact-encoding'
import crypto from 'hypercore-crypto'
import Hyperswarm from 'hyperswarm'
import createTestnet from '@hyperswarm/testnet'
import ProtomuxRPC from 'protomux-rpc'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { RepoStore } from 'atproto-store/lib/repo-store.js'
import { RepoManager } from 'atproto-store/lib/repo-manager.js'
import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
import { keyFromPrivate } from 'atproto-identity'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import { mkDidRouter } from './mk-did-router.js'
import { FleetRelay } from '../lib/fleet-relay.js'
import { pushAnnounce } from './serve-announce.js'
const DID = 'did:plc:routeme'
const HANDLE = 'router.7rnx.net'
const PRIV = new Uint8Array(32).fill(7)
const NOW = '2026-07-03T00:00:00.000Z'
test('PeerRouter: isFronted gate + discover-on-first-use + caching', async (t) => {
const testnet = await createTestnet(3)
const peerDir = mkdtempSync(join(tmpdir(), 'router-peer-'))
const peerStore = openCorestore(peerDir)
const repo = await new RepoStore(peerStore, { name: `atproto-repo:${DID}` }).ready()
let micros = 1_700_000_000_000_000n
const mgr = await new RepoManager(repo, { did: DID, privateKey: PRIV, clock: () => (micros += 1_000_000n) }).init()
const seq = await new FirehoseSequencer(peerStore.get({ name: 'firehose' })).ready()
const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3a', record: { $type: 'app.bsky.feed.post', text: 'routed', createdAt: NOW } })
await seq.sequenceCommit(w, { time: NOW })
const peerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
peerSwarm.on('connection', (conn) => {
peerStore.replicate(conn)
pushAnnounce(conn, { did: DID, handle: HANDLE, repo, seq })
})
const disc = peerSwarm.join(discoveryTopic(DID), { server: true, client: false })
await disc.flushed()
const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'router-')))
const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const relay = await new FleetRelay(routerStore.get({ name: 'peer-relay-fh', valueEncoding: 'json' })).ready()
const router = mkDidRouter(routerSwarm, routerStore, {
fronted: [DID],
resolveVerifyKey: async () => keyFromPrivate(PRIV).publicKey, // injected; no PLC fetch in the test
relay
})
t.teardown(async () => {
await router.close(); await relay.close()
await peerSwarm.destroy(); await routerSwarm.destroy()
await seq.close(); await repo.close(); await peerStore.close(); await routerStore.close()
await testnet.destroy()
rmSync(peerDir, { recursive: true, force: true })
})
t.ok(router.isFronted(DID), 'DID is fronted')
t.absent(router.isFronted('did:plc:other'), 'unknown DID is not fronted')
const front = await router.front(DID)
const { carBytes } = await front.getRepo()
t.ok(carBytes.length > 0, 'router.front(did) discovered + serves the peer repo')
t.is(front.did, DID, 'fronted the right DID')
const again = await router.front(DID)
t.is(again, front, 'front(did) is cached (same instance)')
// the fronted peer's firehose is folded into the aggregate crawl stream
const ac = new AbortController()
const it = relay.subscribe(0, { signal: ac.signal })
const firstFrame = await it.next()
ac.abort()
t.ok(firstFrame.value && firstFrame.value.seq >= 1, 'fronted peer firehose aggregated into the relay (crawl stream)')
await t.exception(router.front('did:plc:other'), 'front() rejects a non-fronted DID')
})
const waitFor = async (fn, { timeout = 25000, interval = 150 } = {}) => {
const end = Date.now() + timeout
while (Date.now() < end) { if (await fn()) return true; await new Promise((r) => setTimeout(r, interval)) }
return false
}
// Regression: a fronted peer's NEW commits must propagate to the bridge — both over the live
// connection AND after the peer restarts (reconnect). The bug: discover() replicated only the
// first connection, so a front froze at first-connect head and new posts never appeared until
// the bridge restarted. Fix: PeerRouter replicates the corestore on every (re)connection.
test('PeerRouter live-follows a fronted peer across new commits AND a peer reconnect', async (t) => {
const testnet = await createTestnet(3)
const DID2 = 'did:plc:followme'
const HANDLE2 = 'follow.7rnx.net'
const PRIV2 = new Uint8Array(32).fill(5)
const T = '2026-07-04T00:00:00.000Z'
const peerDir = mkdtempSync(join(tmpdir(), 'follow-peer-'))
const peerStore = openCorestore(peerDir)
const repo = await new RepoStore(peerStore, { name: `atproto-repo:${DID2}` }).ready()
let micros = 1_700_000_000_000_000n
const mgr = await new RepoManager(repo, { did: DID2, privateKey: PRIV2, clock: () => (micros += 1_000_000n) }).init()
const seq = await new FirehoseSequencer(peerStore.get({ name: 'firehose' })).ready()
const post = async (rkey, text) => {
const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey, record: { $type: 'app.bsky.feed.post', text, createdAt: T } })
await seq.sequenceCommit(w, { time: T })
return w.rev
}
const rev1 = await post('a1', 'first')
// A restarted peer reuses its PERSISTENT swarm keypair (loadOrCreateSwarmKeyPair) so its
// serverKey/DHT identity is stable — that is what lets Hyperswarm reconnect promptly after a
// reboot (a random keypair per boot would leave the front stale). We model that here with a
// fixed keypair across the swarm swap.
const peerSwarmKeyPair = crypto.keyPair(new Uint8Array(32).fill(11))
const mkPeerSwarm = () => {
const s = new Hyperswarm({ bootstrap: testnet.bootstrap, keyPair: peerSwarmKeyPair })
s.on('connection', (conn) => {
peerStore.replicate(conn)
pushAnnounce(conn, { did: DID2, handle: HANDLE2, repo, seq })
})
return s
}
let peerSwarm = mkPeerSwarm()
await peerSwarm.join(discoveryTopic(DID2), { server: true, client: false }).flushed()
const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'follow-router-')))
const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const router = mkDidRouter(routerSwarm, routerStore, { fronted: [DID2], resolveVerifyKey: async () => keyFromPrivate(PRIV2).publicKey })
t.teardown(async () => {
await router.close()
await peerSwarm.destroy(); await routerSwarm.destroy()
await seq.close(); await repo.close(); await peerStore.close(); await routerStore.close()
await testnet.destroy()
rmSync(peerDir, { recursive: true, force: true })
})
const front = await router.front(DID2)
t.is((await front.getLatestCommit()).rev, rev1, 'fronted at the first commit')
// 1) live-follow over the existing connection
const rev2 = await post('a2', 'second')
const followedLive = await waitFor(async () => { const lc = await front.getLatestCommit(); return !!lc && lc.rev === rev2 })
t.ok(followedLive, 'front live-follows a new commit over the live connection')
// 2) the peer RESTARTS (swarm swap), then appends — the front must re-replicate on reconnect
await peerSwarm.destroy()
peerSwarm = mkPeerSwarm()
await peerSwarm.join(discoveryTopic(DID2), { server: true, client: false }).flushed()
const rev3 = await post('a3', 'after reconnect')
const followedReconnect = await waitFor(async () => { const lc = await front.getLatestCommit(); return !!lc && lc.rev === rev3 })
t.ok(followedReconnect, 'front live-follows a new commit AFTER a peer reconnect (persistent replicate)')
})
// DURABILITY regression: a peer can REBUILD its membrane cores (new repoKey/firehoseKey under the same DID
// — e.g. a storage repair). Before the fix, front() cached the PeerFront by DID and never re-checked its
// keys, so it served the DEAD old cores forever until the bridge process restarted. Fix: learn() evicts the
// stale front on a key-change announce so front() re-opens the NEW cores. No restart.
test('PeerRouter re-fronts a peer that REBUILDS its cores with new keys — no restart', async (t) => {
const testnet = await createTestnet(3)
const D = 'did:plc:rebuildme'; const P = new Uint8Array(32).fill(9); const T = '2026-07-05T00:00:00.000Z'
const peerDir = mkdtempSync(join(tmpdir(), 'rebuild-peer-')); const peerStore = openCorestore(peerDir)
let micros = 1_700_000_000_000_000n
const mkCores = async (v) => {
const repo = await new RepoStore(peerStore, { name: `atproto-repo:${D}:${v}` }).ready()
const mgr = await new RepoManager(repo, { did: D, privateKey: P, clock: () => (micros += 1_000_000n) }).init()
const seq = await new FirehoseSequencer(peerStore.get({ name: `firehose:${v}` })).ready()
return { repo, mgr, seq }
}
const post = async (mgr, seq, rkey, text) => { const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey, record: { $type: 'app.bsky.feed.post', text, createdAt: T } }); await seq.sequenceCommit(w, { time: T }); return w.rev }
let cores = await mkCores('v1'); const rev1 = await post(cores.mgr, cores.seq, 'a1', 'v1')
const kp = crypto.keyPair(new Uint8Array(32).fill(13))
const mkSwarm = () => { const s = new Hyperswarm({ bootstrap: testnet.bootstrap, keyPair: kp }); s.on('connection', (conn) => { peerStore.replicate(conn); pushAnnounce(conn, { did: D, handle: 'rebuild.7rnx.net', repo: cores.repo, seq: cores.seq }) }); return s }
let peerSwarm = mkSwarm(); await peerSwarm.join(discoveryTopic(D), { server: true, client: false }).flushed()
const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'rebuild-router-'))); const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const relay = await new FleetRelay(routerStore.get({ name: 'fh', valueEncoding: 'json' })).ready()
const router = mkDidRouter(routerSwarm, routerStore, { fronted: [D], resolveVerifyKey: async () => keyFromPrivate(P).publicKey, relay })
t.teardown(async () => { await router.close(); await relay.close(); await peerSwarm.destroy(); await routerSwarm.destroy(); await cores.seq.close(); await cores.repo.close(); await peerStore.close(); await routerStore.close(); await testnet.destroy(); rmSync(peerDir, { recursive: true, force: true }) })
const f1 = await router.front(D)
t.is((await f1.getLatestCommit()).rev, rev1, 'fronted the v1 cores')
const v1key = b4a.toString(f1._repoKey, 'hex')
// REBUILD: brand-new cores (new keys) under the same DID; peer reconnects announcing them.
await peerSwarm.destroy()
cores = await mkCores('v2'); const rev2 = await post(cores.mgr, cores.seq, 'b1', 'v2 after rebuild')
peerSwarm = mkSwarm(); await peerSwarm.join(discoveryTopic(D), { server: true, client: false }).flushed()
const refronted = await waitFor(async () => {
const f = await router.front(D).catch(() => null)
if (!f || b4a.toString(f._repoKey, 'hex') === v1key) return false
const lc = f.getLatestCommit()
return !!lc && lc.rev === rev2
})
t.ok(refronted, 'router re-fronts with the NEW cores after a rebuild (no restart, new repoKey + rev2)')
})