// CROSS-REPO E2E GATE — the standing pre-flight that would have caught all three prod bugs offline.
//
// Wires the FULL seam that no unit suite crossed: a REAL SwarmDevice (this repo) -> serveMembrane ->
// the REAL bridge PeerFront + FleetRelay (atproto-bridge, imported relatively) -> subscribeRepos.
// Asserts a swarm post emerges on the bridge aggregate as a #commit WITH ops (not #sync-only), that
// it verifies against the DID key, and that after a RESTART the fold stays INCREMENTAL (1 op, no
// full-set re-author). That single assertion covers: membrane restore (this repo), addSource fold +
// per-core dedup (bridge). Runs on an in-process testnet — never touches prod.
import '../lib/bare-globals.js'
import test from 'brittle'
import crypto from 'hypercore-crypto'
import fs from 'bare-fs'
import b4a from 'b4a'
import Hyperswarm from 'hyperswarm'
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 { decodeCommitFrame } from 'atproto-store/lib/firehose.js'
import { loadCommit, readCar } from 'atproto-repo/lib/repo.js'
import { verifyCommit } from 'atproto-repo/lib/commit.js'
import { deriveAtprotoKey } from 'atproto-identity'
import { decodeDidKey } from 'atproto-identity/lib/did-key.js'
// the REAL bridge fold — sibling package, imported relatively for the cross-repo gate
import { PeerFront } from '../../atproto-bridge/pds/lib/peer-front.js'
import { FleetRelay } from '../../atproto-bridge/pds/lib/fleet-relay.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:e2efederation2222222222'
const POST = 'app.bsky.feed.post'
const signKey = deriveAtprotoKey(ROOT)
const { pubkey } = decodeDidKey(signKey.didKey)
const tmp = () => { const d = '/tmp/hpp-e2e-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d }
const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
// Boot a device on FIXED dirs + a FIXED serve keypair so a restart reopens the same state and keeps a
// stable serverKey/firehoseKey (mirrors the persistent-keypair prod fix), letting the bridge reconnect.
async function boot (stateDir, membraneDir, serveSeed, testnet) {
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(stateDir, crypto.hash(b4a.from(DID)), {
roomSeed: crypto.hash(b4a.from('e2e-seed')), encryptionKey: crypto.hash(b4a.from('e2e-enc')), 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.restore() // bootSwarmDevice restores at boot BEFORE the control surface accepts writes
const swarm = new Hyperswarm({ bootstrap: testnet.bootstrap, keyPair: crypto.keyPair(serveSeed) })
const store = openCorestore(membraneDir)
const serving = await serveMembrane(device, { corestore: store, swarm, did: DID, handle: 'e2e.test' })
await serving.ready()
return { A, device, swarm, store, serving }
}
// Pull decoded #commit frames off the bridge aggregate (skip #sync/other; retry as replication catches up).
async function commitsFrom (relay, minCount) {
for (let attempt = 0; attempt < 40; attempt++) {
const ac = new AbortController()
const it = relay.subscribe(0, { signal: ac.signal })
const commits = []
try {
for (let k = 0; k < 50; k++) {
const r = await Promise.race([it.next(), new Promise((res) => setTimeout(() => res({ done: true }), 300))])
if (r.done || !r.value) break
try { const { body } = decodeCommitFrame(r.value.bytes); if (body && Array.isArray(body.ops)) commits.push(body) } catch { /* #sync/other */ }
}
} finally { ac.abort() }
if (commits.length >= minCount) return commits
await sleep(400)
}
return []
}
test('e2e: a swarm post emerges on the bridge aggregate as a #commit with ops, verified + incremental across restart', async (t) => {
const testnet = await createTestnet(3)
const stateDir = tmp(); const membraneDir = tmp(); const brDir = tmp(); const aggDir = tmp()
const serveSeed = crypto.hash(b4a.from('e2e-serve-keypair'))
const cleanup = []
t.teardown(async () => { for (const fn of cleanup.reverse()) { try { await fn() } catch {} } await testnet.destroy() })
// 1. device posts one record + signs (real SwarmDevice -> membrane firehose)
let d = await boot(stateDir, membraneDir, serveSeed, testnet)
cleanup.push(() => d.serving.close(), () => d.swarm.destroy(), () => d.store.close(), () => d.A.close())
await d.device.putRecord(POST, '3aaapost0001', { $type: POST, text: 'hello from the swarm', createdAt: '2026-07-05T00:00:00.000Z' })
await d.device.claimLease(); await d.device.sign()
await d.serving.sync()
// 2. the REAL bridge fronts it keylessly + folds into the aggregate
const brSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }); const brStore = openCorestore(brDir)
cleanup.push(() => brSwarm.destroy(), () => brStore.close())
// replication is the swarm OWNER's job (in prod PeerRouter attaches this so it survives reconnects) —
// PeerFront.discover intentionally does NOT replicate its RPC conn. Without this the cores never sync.
brSwarm.on('connection', (conn) => brStore.replicate(conn))
const front = await PeerFront.discover(brSwarm, brStore, DID, { verifyKey: pubkey, timeout: 20000 })
const relay = await new FleetRelay(openCorestore(aggDir).get({ name: 'agg', valueEncoding: 'json' })).ready()
cleanup.push(() => relay.close())
await relay.addSource(b4a.toString(front.seq.core.key, 'hex'), front.seq.core)
// 3. assert: the post folded as a #commit WITH ops (the #sync-only bug), and verifies keylessly
let commits = await commitsFrom(relay, 1)
t.ok(commits.length >= 1, 'the swarm post folded into the aggregate as a #commit (NOT #sync-only)')
const sc0 = loadCommit(await readCar(commits[commits.length - 1].blocks)).value
t.ok(verifyCommit(sc0, pubkey), 'the folded #commit verifies against the DID key (keyless)')
const paths0 = commits.flatMap((c) => c.ops.map((o) => o.path))
t.ok(paths0.includes(POST + '/3aaapost0001'), 'the aggregate carries the create op for the posted record')
// 4. RESTART the device (new instance, same state + membrane dirs + serve keypair), post again
await d.serving.close(); await d.swarm.destroy(); await d.store.close(); await d.A.close()
d = await boot(stateDir, membraneDir, serveSeed, testnet)
cleanup.push(() => d.serving.close(), () => d.swarm.destroy(), () => d.store.close(), () => d.A.close())
await d.device.claimLease()
await d.device.putRecord(POST, '3bbbpost0002', { $type: POST, text: 'second post after restart', createdAt: '2026-07-05T00:00:01.000Z' })
await d.device.sign()
await d.serving.sync()
// 5. assert: the restart frame is INCREMENTAL (a single op for the new record — not a full-set re-author)
const after = await commitsFrom(relay, 2)
t.ok(after.length >= 2, 'the post-restart commit also folded through the bridge')
const restartFrame = after.find((c) => c.ops.some((o) => o.path === POST + '/3bbbpost0002'))
t.ok(restartFrame, 'the aggregate carries the post-restart record')
t.is(restartFrame.ops.length, 1, 'the post-restart #commit is INCREMENTAL (1 op) — no full-set re-author (the restore fix)')
})