// Phase D slice 2: the LIVE serve/replicate e2e. A swarm device runs serveMembrane; a minimal
// membrane client (mirroring atproto-bridge PeerFront: get-coords -> replicate corestore -> open
// repo+firehose by key -> read) fronts it KEYLESSLY over a testnet swarm. Proves the full wiring,
// not just the projection: the membrane pulls the swarm's firehose + repo and assembles a verified
// CAR without ever holding the signing key.
import '../lib/bare-globals.js'
import test from 'brittle'
import crypto from 'hypercore-crypto'
import fs from 'bare-fs'
import b4a from 'b4a'
import c from 'compact-encoding'
import Hyperswarm from 'hyperswarm'
import Protomux from 'protomux'
import ProtomuxRPC from 'protomux-rpc'
import createTestnet from '@hyperswarm/testnet'
import { StateBaseV3 } from 'atproto-room/lib/state-base.js'
import { SwarmDevice } from '../lib/swarm-device.js'
import { serveMembrane } from 'atproto-room/lib/serve-membrane.js'
import { makeAttest, makeVerifyAttest, attestDeviceChain, deviceKeetKeyPair, rootIdentityPublicKey } from 'atproto-room/lib/membership.js'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { RepoStore } from 'atproto-store/lib/repo-store.js'
import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
import { decodeCommitFrame } from 'atproto-store/lib/firehose.js'
import { getRepoFromCommit } from 'atproto-store/lib/read-view.js'
import { readCar, loadCommit, listRecords } from 'atproto-repo/lib/repo.js'
import { verifyCommit } from 'atproto-repo/lib/commit.js'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import { deriveAtprotoKey } from 'atproto-identity'
import { decodeDidKey } from 'atproto-identity/lib/did-key.js'
const ROOT = 'abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about'
const DEVA = 'legal winner thank year wave sausage worth useful legal winner thank yellow'
const DID = 'did:plc:swarmserve2222222222222'
const POST = 'app.bsky.feed.post'
const RPC_PROTOCOL = 'hyper-proto-peer/rpc/v1'
const signKey = deriveAtprotoKey(ROOT)
const { pubkey } = decodeDidKey(signKey.didKey)
const tmp = () => { const d = '/tmp/hpp-ms-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d }
const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
test('the keyless membrane fronts a serving swarm device over the swarm (announce -> replicate -> verify)', async (t) => {
const testnet = await createTestnet(3)
const rootPub = await rootIdentityPublicKey(ROOT)
const dev = await deviceKeetKeyPair(DEVA); const proof = await attestDeviceChain({ rootMnemonic: ROOT, devicePublicKey: dev.publicKey })
const members = [{ key: b4a.toString(dev.publicKey, 'hex'), role: 1 }]
const A = new StateBaseV3(tmp(), crypto.randomBytes(32), {
roomSeed: crypto.randomBytes(32), encryptionKey: crypto.randomBytes(32), network: false,
selfKey: dev.publicKey, attest: makeAttest(dev, proof), verifyAttest: makeVerifyAttest(rootPub),
getMembers: async () => members
})
await A.ready(); await A.projectMembers(members)
const device = new SwarmDevice(A, { did: DID, privateKey: signKey.privateKey, self: b4a.toString(dev.publicKey, 'hex'), firehose: true })
await device.putRecord(POST, '3aaa', { $type: POST, text: 'hello bsky', createdAt: '2026-07-04T00:00:00.000Z' })
await device.claimLease(); await device.sign()
// the swarm device serves its membrane surface over testnet
const serveSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }); const serveStore = openCorestore(tmp())
const serving = await serveMembrane(device, { corestore: serveStore, swarm: serveSwarm, did: DID, handle: 'swarmserve.test' })
await serving.ready()
const clientSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }); const clientStore = openCorestore(tmp())
t.teardown(async () => {
serving.close(); await serveSwarm.destroy(); await clientSwarm.destroy()
try { await serveStore.close(); await clientStore.close(); await A.close() } catch {}
await testnet.destroy()
})
// membrane client: connect on the DID topic, replicate the corestore, receive the pushed announce
// (mirrors atproto-bridge PeerRouter._identify — the SOLE discovery path, no get-coords).
const coords = await new Promise((resolve) => {
clientSwarm.on('connection', (conn) => {
clientStore.replicate(conn)
const achan = Protomux.from(conn).createChannel({ protocol: 'atproto/did-announce/v1' })
if (!achan) return
achan.addMessage({ encoding: c.json, onmessage: (a) => resolve(a) })
achan.open()
})
clientSwarm.join(discoveryTopic(DID), { client: true, server: false })
})
t.is(coords.did, DID, 'announce carries the DID')
t.ok(coords.repoKey && coords.firehoseKey, 'announce carries repo + firehose Hypercore keys')
const repo = await new RepoStore(clientStore, { key: coords.repoKey }).ready()
const seq = await new FirehoseSequencer(clientStore.get({ key: coords.firehoseKey })).ready()
for (let i = 0; i < 40 && seq.core.length === 0; i++) { await seq.core.update().catch(() => {}); if (seq.core.length === 0) await sleep(500) }
t.ok(seq.core.length > 0, 'the firehose replicated to the membrane client')
const { body } = decodeCommitFrame(await seq.core.get(seq.core.length - 1))
const signedCommit = loadCommit(await readCar(body.blocks)).value
t.ok(verifyCommit(signedCommit, pubkey), 'the replicated #commit verifies against the DID key (keyless)')
let recs = []
for (let i = 0; i < 40 && recs.length === 0; i++) { await repo.core.update().catch(() => {}); recs = await repo.records().catch(() => []); if (!recs.length) await sleep(500) }
const { carBytes } = await getRepoFromCommit(repo, signedCommit)
const keys = listRecords(await readCar(carBytes)).map((r) => r.key)
t.alike(keys, ['app.bsky.feed.post/3aaa'], 'the membrane assembled the repo CAR from the replicated swarm — keylessly')
})
// Corrected discovery: serve-membrane PUSHES an announce (atproto/did-announce/v1) carrying the core
// keys + lengths, so a consumer learns them WITHOUT a get-coords round-trip (no serverKey directionality).
test('serve-membrane PUSHES an announce — consumer learns the cores WITHOUT get-coords', async (t) => {
const testnet = await createTestnet(3)
const rootPub = await rootIdentityPublicKey(ROOT)
const dev = await deviceKeetKeyPair(DEVA); const proof = await attestDeviceChain({ rootMnemonic: ROOT, devicePublicKey: dev.publicKey })
const members = [{ key: b4a.toString(dev.publicKey, 'hex'), role: 1 }]
const A = new StateBaseV3(tmp(), crypto.randomBytes(32), {
roomSeed: crypto.randomBytes(32), encryptionKey: crypto.randomBytes(32), network: false,
selfKey: dev.publicKey, attest: makeAttest(dev, proof), verifyAttest: makeVerifyAttest(rootPub),
getMembers: async () => members
})
await A.ready(); await A.projectMembers(members)
const device = new SwarmDevice(A, { did: DID, privateKey: signKey.privateKey, self: b4a.toString(dev.publicKey, 'hex'), firehose: true })
await device.putRecord(POST, '3aaa', { $type: POST, text: 'announce me', createdAt: '2026-07-13T00:00:00.000Z' })
await device.claimLease(); await device.sign()
const serveSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }); const serveStore = openCorestore(tmp())
const serving = await serveMembrane(device, { corestore: serveStore, swarm: serveSwarm, did: DID, handle: 'announce.test' })
await serving.ready()
const clientSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }); const clientStore = openCorestore(tmp())
t.teardown(async () => {
serving.close(); await serveSwarm.destroy(); await clientSwarm.destroy()
try { await serveStore.close(); await clientStore.close(); await A.close() } catch {}
await testnet.destroy()
})
// consumer: open ONLY the announce channel — never call get-coords — and receive the pushed announce.
const announce = await new Promise((resolve, reject) => {
const to = setTimeout(() => reject(new Error('no announce received')), 15000)
clientSwarm.on('connection', (conn) => {
clientStore.replicate(conn)
const chan = Protomux.from(conn).createChannel({ protocol: 'atproto/did-announce/v1' })
if (!chan) return
const msg = chan.addMessage({ encoding: c.json, onmessage: (a) => { if (a && a.did === DID) { clearTimeout(to); resolve(a) } } })
chan.open()
})
clientSwarm.join(discoveryTopic(DID), { client: true, server: true })
})
t.is(announce.did, DID, 'announce carries the DID (push, no get-coords request)')
t.ok(announce.repoKey && announce.firehoseKey, 'announce PUSHES repo + firehose core keys')
t.ok(announce.firehoseLength >= 1, 'announce carries the firehose length hint')
})
// STORAGE-LEAK REGRESSION: the sign loop calls sync() every ~2s. It MUST be a no-op when the __commit__
// log has not advanced. Before the gate, projectRepo re-put ALL records into the append-only Hyperbee repo
// core on EVERY sync — even with zero new commits — growing it ~one record-set / tick → 1.3M blocks / GB in
// hours (a runaway storage+CPU leak; favela then re-replicated the bloated core at ~100 MB/s).
test('sync() is a no-op without new commits — no repo-core balloon (leak regression)', async (t) => {
const testnet = await createTestnet(3)
const rootPub = await rootIdentityPublicKey(ROOT)
const dev = await deviceKeetKeyPair(DEVA); const proof = await attestDeviceChain({ rootMnemonic: ROOT, devicePublicKey: dev.publicKey })
const members = [{ key: b4a.toString(dev.publicKey, 'hex'), role: 1 }]
const A = new StateBaseV3(tmp(), crypto.randomBytes(32), {
roomSeed: crypto.randomBytes(32), encryptionKey: crypto.randomBytes(32), network: false,
selfKey: dev.publicKey, attest: makeAttest(dev, proof), verifyAttest: makeVerifyAttest(rootPub),
getMembers: async () => members
})
await A.ready(); await A.projectMembers(members)
const device = new SwarmDevice(A, { did: DID, privateKey: signKey.privateKey, self: b4a.toString(dev.publicKey, 'hex'), firehose: true })
await device.putRecord(POST, '3aaa', { $type: POST, text: 'one', createdAt: '2026-07-17T00:00:00.000Z' })
await device.claimLease(); await device.sign()
const serveSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }); const serveStore = openCorestore(tmp())
const serving = await serveMembrane(device, { corestore: serveStore, swarm: serveSwarm, did: DID, handle: 'leak.test' })
await serving.ready()
t.teardown(async () => { serving.close(); await serveSwarm.destroy(); try { await serveStore.close(); await A.close() } catch {} ; await testnet.destroy() })
const len0 = serving.repo.core.length
for (let i = 0; i < 25; i++) await serving.sync() // 25 sign-loop ticks, NO new records
t.is(serving.repo.core.length, len0, '25 no-op syncs do NOT grow the repo core (was +blocks/tick → the 1.3M-block leak)')
// a genuine new record STILL projects — the gate does not starve real writes
await device.putRecord(POST, '3bbb', { $type: POST, text: 'two', createdAt: '2026-07-17T00:01:00.000Z' })
await device.sign()
await serving.sync()
t.ok(serving.repo.core.length > len0, 'a real new commit DOES project (records still flow through the gate)')
})