/**
* Multi-round write test — isolates whether subsequent writes
* propagate between two peers after initial bidirectional sync.
*
* Tests with piped corestores (no Hyperswarm) to rule out
* network/wakeup issues.
*/
import test from 'brittle'
import Corestore from 'corestore'
import tmp from 'test-tmp'
import path from 'bare-path'
import { LinkedRepo } from '../../lib/repo-link/linked-repo.js'
import { IdentityRegistry } from '../../lib/auth/identity-registry.js'
function replicate (storeA, storeB) {
const s1 = storeA.replicate(true)
const s2 = storeB.replicate(false)
s1.on('error', () => {})
s2.on('error', () => {})
s1.pipe(s2).pipe(s1)
return () => {
s1.destroy()
s2.destroy()
return Promise.all([
new Promise(r => s1.on('close', r)),
new Promise(r => s2.on('close', r))
])
}
}
function eventFlush () {
return new Promise(resolve => setImmediate(resolve))
}
async function syncBases (bases, retries = 20) {
for (let i = 0; i < retries; i++) {
for (const base of bases) await base._base.update()
await eventFlush()
}
}
// ── Test: LinkedRepo multi-round with piped corestores ───────
test('linked repo: 3 rounds of bidirectional writes (piped)', async (t) => {
const dirA = await tmp(t)
const dirB = await tmp(t)
const storeA = new Corestore(path.join(dirA, 'cores'), { primaryKey: Buffer.alloc(32, 0), unsafe: true })
const storeB = new Corestore(path.join(dirB, 'cores'), { primaryKey: Buffer.alloc(32, 1), unsafe: true })
t.teardown(() => Promise.all([storeA.close(), storeB.close()]), { order: 2 })
const linkedA = new LinkedRepo({ corestore: storeA, bootstrapKey: null, repoName: 'test' })
await linkedA.ready()
const linkedB = new LinkedRepo({ corestore: storeB, bootstrapKey: linkedA.bootstrapKeyHex, repoName: 'test' })
await linkedB.ready()
// Add B as writer (owner = indexer)
const close = replicate(storeA, storeB)
t.teardown(close, { order: 1 })
await linkedA.addWriter('b', linkedB.localKey.toString('hex'), null, null, 'owner')
await syncBases([linkedA, linkedB])
// Round 1
await linkedA.createIssue('Round 1 from A', '', 'alice')
await syncBases([linkedA, linkedB])
let bIssues = await linkedB.listIssues()
t.is(bIssues.length, 1, 'R1: B sees A\'s issue')
await linkedB.createIssue('Round 1 from B', '', 'bob')
await syncBases([linkedA, linkedB])
let aIssues = await linkedA.listIssues()
t.is(aIssues.length, 2, 'R1: A sees both issues')
// Round 2
await linkedA.createIssue('Round 2 from A', '', 'alice')
await syncBases([linkedA, linkedB])
bIssues = await linkedB.listIssues()
t.is(bIssues.length, 3, 'R2: B sees 3 issues')
await linkedB.createIssue('Round 2 from B', '', 'bob')
await syncBases([linkedA, linkedB])
aIssues = await linkedA.listIssues()
t.is(aIssues.length, 4, 'R2: A sees 4 issues')
// Round 3
await linkedA.createIssue('Round 3 from A', '', 'alice')
await linkedB.createIssue('Round 3 from B', '', 'bob')
await syncBases([linkedA, linkedB])
aIssues = await linkedA.listIssues()
bIssues = await linkedB.listIssues()
t.is(aIssues.length, 6, 'R3: A sees all 6')
t.is(bIssues.length, 6, 'R3: B sees all 6')
t.is(aIssues.length, bIssues.length, 'R3: both converged')
})
// ── Test: Identity registry multi-round (piped) ─────────────
test('identity: 3 rounds of bidirectional writes (piped)', async (t) => {
const dirA = await tmp(t)
const dirB = await tmp(t)
const storeA = new Corestore(path.join(dirA, 'cores'), { primaryKey: Buffer.alloc(32, 0), unsafe: true })
const storeB = new Corestore(path.join(dirB, 'cores'), { primaryKey: Buffer.alloc(32, 1), unsafe: true })
t.teardown(() => Promise.all([storeA.close(), storeB.close()]), { order: 2 })
const regA = new IdentityRegistry({ corestore: storeA })
await regA.ready()
const regB = new IdentityRegistry({ corestore: storeB, bootstrapKey: regA.bootstrapKeyHex })
await regB.ready()
const close = replicate(storeA, storeB)
t.teardown(close, { order: 1 })
await regA.addWriter(regB.localKey)
await syncBases([regA, regB])
// Round 1
await regA.createUser({ name: 'alice-1', permissions: ['admin'] })
await syncBases([regA, regB])
t.is((await regB.listAllUsers()).length, 1, 'R1: B sees 1')
await regB.createUser({ name: 'bob-1', permissions: ['write'] })
await syncBases([regA, regB])
t.is((await regA.listAllUsers()).length, 2, 'R1: A sees 2')
// Round 2
await regA.createUser({ name: 'alice-2', permissions: ['write'] })
await syncBases([regA, regB])
t.is((await regB.listAllUsers()).length, 3, 'R2: B sees 3')
await regB.createUser({ name: 'bob-2', permissions: ['write'] })
await syncBases([regA, regB])
t.is((await regA.listAllUsers()).length, 4, 'R2: A sees 4')
// Round 3
await regA.createUser({ name: 'alice-3', permissions: ['write'] })
await regB.createUser({ name: 'bob-3', permissions: ['write'] })
await syncBases([regA, regB])
const finalA = await regA.listAllUsers()
const finalB = await regB.listAllUsers()
t.is(finalA.length, 6, 'R3: A sees all 6')
t.is(finalB.length, 6, 'R3: B sees all 6')
})
// ── Test: LinkedRepo multi-round over Hyperswarm testnet ─────
test('linked repo: 3 rounds over testnet (with wakeup)', async (t) => {
const createTestnet = (await import('hyperdht/testnet')).default
const Hyperswarm = (await import('hyperswarm')).default
const testnet = await createTestnet(3, t.teardown)
const dirA = await tmp(t)
const dirB = await tmp(t)
const storeA = new Corestore(path.join(dirA, 'cores'), { primaryKey: Buffer.alloc(32, 10), unsafe: true })
const storeB = new Corestore(path.join(dirB, 'cores'), { primaryKey: Buffer.alloc(32, 11), unsafe: true })
t.teardown(() => Promise.all([storeA.close(), storeB.close()]), { order: 2 })
const linkedA = new LinkedRepo({ corestore: storeA, bootstrapKey: null, repoName: 'test' })
await linkedA.ready()
const linkedB = new LinkedRepo({ corestore: storeB, bootstrapKey: linkedA.bootstrapKeyHex, repoName: 'test' })
await linkedB.ready()
// Add B as writer
await linkedA.addWriter('b', linkedB.localKey.toString('hex'), null, null, 'owner')
const swarmA = new Hyperswarm({ bootstrap: testnet.bootstrap })
const swarmB = new Hyperswarm({ bootstrap: testnet.bootstrap })
t.teardown(async () => {
await swarmA.destroy()
await swarmB.destroy()
})
swarmA.on('connection', conn => {
const stream = storeA.replicate(conn)
// Add wakeup protocol
linkedA._base.wakeupProtocol.addStream(stream)
})
swarmB.on('connection', conn => {
const stream = storeB.replicate(conn)
linkedB._base.wakeupProtocol.addStream(stream)
})
// Both join the LinkedRepo's discovery key
const topic = linkedA._base.discoveryKey
await swarmA.join(topic, { server: true, client: true }).flushed()
swarmB.join(topic, { server: true, client: true })
await swarmB.flush()
// Wait for connection + initial sync
await new Promise(r => setTimeout(r, 2000))
await syncBases([linkedA, linkedB])
// Round 1
await linkedA.createIssue('R1 from A', '', 'alice')
await syncBases([linkedA, linkedB])
t.is((await linkedB.listIssues()).length, 1, 'R1: B sees 1')
await linkedB.createIssue('R1 from B', '', 'bob')
await syncBases([linkedA, linkedB])
t.is((await linkedA.listIssues()).length, 2, 'R1: A sees 2')
// Round 2
await linkedA.createIssue('R2 from A', '', 'alice')
await syncBases([linkedA, linkedB])
t.is((await linkedB.listIssues()).length, 3, 'R2: B sees 3')
await linkedB.createIssue('R2 from B', '', 'bob')
await syncBases([linkedA, linkedB])
t.is((await linkedA.listIssues()).length, 4, 'R2: A sees 4')
// Round 3 — sequential (not simultaneous)
await linkedA.createIssue('R3 from A', '', 'alice')
await syncBases([linkedA, linkedB])
t.is((await linkedB.listIssues()).length, 5, 'R3: B sees 5')
await linkedB.createIssue('R3 from B', '', 'bob')
await syncBases([linkedA, linkedB])
t.is((await linkedA.listIssues()).length, 6, 'R3: A sees all 6')
t.is((await linkedB.listIssues()).length, 6, 'R3: B sees all 6')
})
// ── Test: Namespace registry multi-round (piped) ─────────────
test('namespace: 3 rounds of bidirectional writes (piped)', async (t) => {
const { NamespaceRegistry } = await import('../../lib/namespace-registry/namespace-registry.js')
const dirA = await tmp(t)
const dirB = await tmp(t)
const storeA = new Corestore(path.join(dirA, 'cores'), { primaryKey: Buffer.alloc(32, 20), unsafe: true })
const storeB = new Corestore(path.join(dirB, 'cores'), { primaryKey: Buffer.alloc(32, 21), unsafe: true })
t.teardown(() => Promise.all([storeA.close(), storeB.close()]), { order: 2 })
const nsA = new NamespaceRegistry({ corestore: storeA, bootstrapKey: null })
await nsA.ready()
const nsB = new NamespaceRegistry({ corestore: storeB, bootstrapKey: nsA.bootstrapKeyHex })
await nsB.ready()
const close = replicate(storeA, storeB)
t.teardown(close, { order: 1 })
// Add B as owner (indexer) — pass localKey as serverKey
await nsA.addWriter('peer-b', nsB.localKey.toString('hex'), 'owner')
await syncBases([nsA, nsB])
// Round 1
await nsA.create('group-a1', 'group', { displayName: 'A Group 1' })
await syncBases([nsA, nsB])
t.is((await nsB.list()).length, 1, 'R1: B sees 1 group')
await nsB.create('group-b1', 'group', { displayName: 'B Group 1' })
await syncBases([nsA, nsB])
t.is((await nsA.list()).length, 2, 'R1: A sees 2 groups')
// Round 2: add repos to groups
await nsA.addRepo('group-a1', 'pear-git')
await nsB.addRepo('group-b1', 'wirecore')
await syncBases([nsA, nsB])
const gA1onB = await nsB.get('group-a1')
t.ok(gA1onB.repos.includes('pear-git'), 'R2: B sees pear-git in A\'s group')
const gB1onA = await nsA.get('group-b1')
t.ok(gB1onA.repos.includes('wirecore'), 'R2: A sees wirecore in B\'s group')
// Round 3: both create more groups
await nsA.create('group-a2', 'group', { displayName: 'A Group 2' })
await nsB.create('group-b2', 'group', { displayName: 'B Group 2' })
await syncBases([nsA, nsB])
const finalA = await nsA.list()
const finalB = await nsB.list()
t.is(finalA.length, 4, 'R3: A sees all 4 groups')
t.is(finalB.length, 4, 'R3: B sees all 4 groups')
})
test('cleanup', (t) => {
t.pass('done')
t.teardown(() => {
const exit = typeof Bare !== 'undefined' ? Bare.exit : process.exit
if (exit) setTimeout(() => exit(0), 500)
})
})