// Step 3 read-side: the bridge fronts a REMOTE hyper-proto-peer KEYLESSLY. A served peer (repo +
// its own signed firehose, over Hyperswarm) is projected by a PeerFront that holds NO key —
// getRepo assembles a signed CAR from the replicated records + the firehose commit, and
// subscribeRepos streams the peer's signed frames. The signature travels; the key never does.
import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import Hyperswarm from 'hyperswarm'
import createTestnet from '@hyperswarm/testnet'
import c from 'compact-encoding'
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 { readCar, loadCommit } from 'atproto-repo/lib/repo.js'
import { verifyCommit } from 'atproto-repo/lib/commit.js'
import { keyFromPrivate } from 'atproto-identity'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import b4a from 'b4a'
import { serveReplication, connectReplica } from '../lib/fleet-replicate.js'
import { DidPeer, RPC_PROTOCOL } from 'atproto-did-peer'
import { mkDidRouter } from './mk-did-router.js'
import { pushAnnounce } from './serve-announce.js'
const DID = 'did:plc:frontme'
const HANDLE = 'peer.7rnx.net'
const PRIV = new Uint8Array(32).fill(7)
const NOW = '2026-07-03T00:00:00.000Z'
async function waitFor (fn, timeout = 12000) {
const end = Date.now() + timeout
while (Date.now() < end) { if (await fn()) return true; await new Promise(r => setTimeout(r, 100)) }
return false
}
// Test harness: a standalone DidPeer front, for the two read/write-surface tests. Connect to the peer by
// its swarm key, replicate, open the repo+firehose cores via provide() (which seeds + reconciles the head),
// and open the write RPC — the same steps the router does per connection, but for one peer without the full
// DidPeerRouter. Uses connectReplica (a live helper in a separate subsystem) purely as test plumbing.
// Does NOT wait for a head: test 2's peer has no commit yet (it signs on the relayed applyWrite).
async function connectFront (corestore, swarm, serverKeyHex, { repoKey, firehoseKey, did, handle, verifyKey }) {
const conn = await connectReplica(corestore, swarm, serverKeyHex)
const front = new DidPeer(corestore, { did, verifyKey, handle })
front.rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json })
await front.provide({ repoKey, firehoseKey })
return front
}
test('bridge PeerFront serves a remote peer getRepo + subscribeRepos — KEYLESS', async (t) => {
const testnet = await createTestnet(3)
const peerDir = mkdtempSync(join(tmpdir(), 'pf-peer-'))
const bridgeDir = mkdtempSync(join(tmpdir(), 'pf-bridge-'))
// --- peer side: repo + its own signed firehose, served over Hyperswarm ---
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: 'fronted', createdAt: NOW } })
await seq.sequenceCommit(w, { time: NOW })
const peerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const served = serveReplication(peerStore, peerSwarm)
await served.ready()
// --- bridge side: PeerFront holds NO key ---
const bridgeStore = openCorestore(bridgeDir)
const bridgeSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const front = await connectFront(bridgeStore, bridgeSwarm, served.serverKey, { repoKey: repo.core.key, firehoseKey: seq.core.key, did: DID, handle: HANDLE, verifyKey: keyFromPrivate(PRIV).publicKey })
t.teardown(async () => {
served.close()
await peerSwarm.destroy(); await bridgeSwarm.destroy()
await front.close(); await seq.close(); await repo.close()
await peerStore.close(); await bridgeStore.close()
await testnet.destroy()
rmSync(peerDir, { recursive: true, force: true }); rmSync(bridgeDir, { recursive: true, force: true })
})
// getRepo: wait for replication, then assemble the signed CAR keylessly
t.ok(await waitFor(async () => { try { await front.getRepo(); return true } catch { return false } }), 'peer repo+firehose replicated to the bridge')
const { carBytes } = await front.getRepo()
const signed = loadCommit(await readCar(carBytes)).value
const pub = keyFromPrivate(PRIV).publicKey
t.ok(verifyCommit(signed, pub), 'bridge serves a signed getRepo CAR for the remote peer — KEYLESS')
t.is(signed.did, DID, 'CAR commit authored under the peer DID')
// subscribeRepos: the bridge streams the peer's own signed firehose frame
const ac = new AbortController()
const it = front.subscribeRepos(0, { signal: ac.signal })
const first = await it.next()
ac.abort()
t.is(first.value && first.value.seq, 1, 'subscribeRepos yields the peer firehose frame at seq 1')
// keyless read surface (from the replicated repo/firehose)
const rec = await front.getRecord('app.bsky.feed.post', '3a')
t.is(rec && rec.value.text, 'fronted', 'getRecord returns the peer record')
const { records: list } = await front.listRecords('app.bsky.feed.post')
t.ok(list.some(r => r.value.text === 'fronted'), 'listRecords includes the record')
const latest = await front.getLatestCommit()
t.ok(latest && latest.cid && latest.rev, 'getLatestCommit returns { cid, rev }')
t.is(front.getRepoStatus().active, true, 'getRepoStatus reports active')
})
test('PeerFront relays a write over RPC — the peer signs, the bridge stays keyless', async (t) => {
const testnet = await createTestnet(3)
const peerDir = mkdtempSync(join(tmpdir(), 'pfw-peer-'))
const bridgeDir = mkdtempSync(join(tmpdir(), 'pfw-bridge-'))
// peer side: repo + signed firehose + a signing apply-write RPC responder (what hyper-proto-peer's
// serve does). The PEER holds the key and signs; here it authorizes any caller for the test.
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 peerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const served = serveReplication(peerStore, peerSwarm)
peerSwarm.on('connection', (conn) => {
const rpc = new ProtomuxRPC(conn, { protocol: 'hyper-proto-peer/rpc/v1', valueEncoding: c.json })
rpc.respond('apply-write', async (op) => {
const w = await mgr.applyWrite(op)
await seq.sequenceCommit(w, { time: NOW })
return { did: DID, commitCid: w.commitCid.toString(), rev: w.rev }
})
})
await served.ready()
const bridgeStore = openCorestore(bridgeDir)
const bridgeSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const front = await connectFront(bridgeStore, bridgeSwarm, served.serverKey, { repoKey: repo.core.key, firehoseKey: seq.core.key, did: DID, handle: HANDLE, verifyKey: keyFromPrivate(PRIV).publicKey })
t.teardown(async () => {
served.close()
await peerSwarm.destroy(); await bridgeSwarm.destroy()
await front.close(); await seq.close(); await repo.close()
await peerStore.close(); await bridgeStore.close()
await testnet.destroy()
rmSync(peerDir, { recursive: true, force: true }); rmSync(bridgeDir, { recursive: true, force: true })
})
// bridge relays a write; the PEER signs it (the bridge never holds PRIV)
const res = await front.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3b', record: { $type: 'app.bsky.feed.post', text: 'written via the bridge', createdAt: NOW } })
t.ok(res.commitCid, 'peer returned a signed commit for the bridge-relayed write')
// the signed write flows back on the peer's firehose; the bridge keylessly serves the updated repo
t.ok(await waitFor(async () => front.head && (await front.repo.records()).length === 1), 'bridge replicated the peer-signed record')
const { carBytes } = await front.getRepo()
const signed = loadCommit(await readCar(carBytes)).value
const pub = keyFromPrivate(PRIV).publicKey
t.ok(verifyCommit(signed, pub), 'bridge-served CAR (post-write) verifies against the peer key — bridge stayed keyless')
})
// A minimal serve-membrane-equivalent peer: a signed repo + firehose, replicated over its own swarm,
// PUSHING an atproto/did-announce/v1 announce on connect + joining discoveryTopic(did) SYMMETRICALLY.
// Close enough to hyper-proto-peer serveMembrane to exercise the bridge's PeerRouter front path.
async function servePeer (testnet, did, priv) {
const store = openCorestore(mkdtempSync(join(tmpdir(), 'pr-peer-')))
const repo = await new RepoStore(store, { name: `atproto-repo:${did}` }).ready()
let micros = 1_700_000_000_000_000n
const mgr = await new RepoManager(repo, { did, privateKey: priv, clock: () => (micros += 1_000_000n) }).init()
const seq = await new FirehoseSequencer(store.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: did, createdAt: NOW } })
await seq.sequenceCommit(w, { time: NOW })
const sw = new Hyperswarm({ bootstrap: testnet.bootstrap })
sw.on('connection', (conn) => {
store.replicate(conn)
pushAnnounce(conn, { did, handle: `${did.slice(-6)}.test`, repo, seq })
})
await sw.join(discoveryTopic(did), { server: true, client: true }).flushed()
return { store, repo, seq, sw, close: async () => { await sw.destroy(); await seq.close(); await repo.close(); await store.close() } }
}
// The bridge side: a real PeerRouter over one swarm/corestore, fronting a set of DIDs, with an
// injected verify-key resolver (keyByDid) so we don't hit the live PLC directory.
function mkRouter (testnet, fronted, keyByDid) {
const store = openCorestore(mkdtempSync(join(tmpdir(), 'pr-bridge-')))
const swarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const router = mkDidRouter(swarm, store, { fronted, resolveVerifyKey: async (did) => keyByDid[did] })
return { store, swarm, router, close: async () => { await router.close(); await swarm.destroy(); await store.close() } }
}
test('PeerRouter fronts a peer discovered by DID + rejects a wrong key (connect-time auth)', async (t) => {
const testnet = await createTestnet(3)
const peer = await servePeer(testnet, DID, PRIV)
const goodKey = keyFromPrivate(PRIV).publicKey
const wrongKey = keyFromPrivate(new Uint8Array(32).fill(9)).publicKey
const good = mkRouter(testnet, [DID], { [DID]: goodKey })
const bad = mkRouter(testnet, [DID], { [DID]: wrongKey })
t.teardown(async () => { await peer.close(); await good.close(); await bad.close(); await testnet.destroy() })
const front = await good.router.front(DID)
t.is(front.did, DID, 'fronted the peer discovered by DID')
const { carBytes } = await front.getRepo()
t.ok(verifyCommit(loadCommit(await readCar(carBytes)).value, goodKey), 'discovered peer serves a signed CAR (keyless)')
// a squatter/wrong-key: connect-time auth refuses to front
await t.exception(bad.router.front(DID, { timeout: 3000 }), 'refuses to front a repo that does not verify against the DID key')
})
// Regression for the discover RACE (2026-07-06): the old discover used swarm.once('connection'),
// which fires only on a NEW connection — but the router keeps a PERSISTENT connection handler and
// identifies each peer on connect, so the peer is usually ALREADY connected when a front is requested
// ("timeout finding" even though it sits in swarm.connections). front() is now a LOOKUP over the
// router's identified connections. This test waits until the peer is identified, THEN fronts.
test('PeerRouter fronts a peer that is ALREADY connected (front is a lookup, not a probe)', async (t) => {
const testnet = await createTestnet(3)
const peer = await servePeer(testnet, DID, PRIV)
const b = mkRouter(testnet, [DID], { [DID]: keyFromPrivate(PRIV).publicKey })
t.teardown(async () => { await peer.close(); await b.close(); await testnet.destroy() })
// the router joined the topic in its constructor; wait until it has IDENTIFIED the peer's
// connection (announce received + cached), THEN front — exercising the already-connected path.
t.ok(await waitFor(() => b.router._byDid.has(DID)), 'router identified the peer connection on connect')
const front = await b.router.front(DID)
t.is(front.did, DID, 'front resolved from the already-identified connection (no new connection event)')
})
// THE multi-room case (2026-07-06): one membrane, ONE swarm, MULTIPLE DIDs — the keet model
// (one rootCorestore.replicate per connection; each room muxes over it). front() fronting a 2nd DID
// must NOT open a duplicate RPC channel on the 1st peer's stream. The old probe-discover did exactly
// that (DUPLICATE_CHANNEL → boot crash-loop, port never opened). One RPC per connection can't.
test('PeerRouter fronts MULTIPLE DIDs over ONE swarm — no DUPLICATE_CHANNEL (the multi-room case)', async (t) => {
const testnet = await createTestnet(3)
const DID_A = 'did:plc:frontaaaaaaaaaaaaaaaaaa'
const DID_B = 'did:plc:frontbbbbbbbbbbbbbbbbbb'
const PRIV_A = new Uint8Array(32).fill(3)
const PRIV_B = new Uint8Array(32).fill(4)
const keyA = keyFromPrivate(PRIV_A).publicKey
const keyB = keyFromPrivate(PRIV_B).publicKey
const A = await servePeer(testnet, DID_A, PRIV_A)
const B = await servePeer(testnet, DID_B, PRIV_B)
const b = mkRouter(testnet, [DID_A, DID_B], { [DID_A]: keyA, [DID_B]: keyB })
t.teardown(async () => { await A.close(); await B.close(); await b.close(); await testnet.destroy() })
// front BOTH DIDs over the ONE router/swarm (the pds-serve boot order). One RPC per connection ⇒
// fronting the 2nd never opens a second channel on the 1st's stream (the crash we fixed).
const frontA = await b.router.front(DID_A)
const frontB = await b.router.front(DID_B)
t.is(frontA.did, DID_A, 'fronted the first DID')
t.is(frontB.did, DID_B, 'fronted the SECOND DID over the same swarm — no channel collision')
// each fronts its OWN signed repo, keylessly
t.ok(verifyCommit(loadCommit(await readCar((await frontA.getRepo()).carBytes)).value, keyA), 'DID_A CAR verifies against DID_A key')
t.ok(verifyCommit(loadCommit(await readCar((await frontB.getRepo()).carBytes)).value, keyB), 'DID_B CAR verifies against DID_B key')
})
// Regression guard for the firewalled↔firewalled holepunch fix (2026-07-06). The bridge host and
// the self-custody peer are BOTH typically NAT'd; hyperdht only completes a holepunch between two
// firewalled members when BOTH join client:true (a client-only member never punches outbound, so
// its NAT never opens). Proven live: symmetric {server,client} connects in ~2s; a client-only
// bridge logs cAtt>0 but cOpen:0 and never connects. Can't reproduce a real holepunch on a
// localhost testnet, so assert the load-bearing join OPTIONS via a spy swarm (no connection ever
// fires → discover rejects on its own short timeout; we only need the synchronous join opts).
test('PeerRouter joins each fronted DID topic symmetrically (server:true is load-bearing for NAT holepunch)', (t) => {
const joins = []
const fakeSwarm = {
on () {}, off () {}, connections: [],
join (topic, opts) { joins.push(opts); return { flushed: async () => {} } },
flush () { return Promise.resolve() }
}
// the router joins every fronted DID's topic in its constructor (like keet activateRoom)
const r = mkDidRouter(fakeSwarm, {}, { fronted: [DID], resolveVerifyKey: async () => new Uint8Array(33).fill(1) })
t.ok(r, 'router constructed')
t.is(joins.length, 1, 'router joins the fronted DID topic once')
t.is(joins[0].client, true, 'joins as client (look up)')
t.is(joins[0].server, true, 'joins as server — REQUIRED so a firewalled bridge punches outbound to a firewalled peer')
})