atproto-bridge / pds / test / fleet-replicate.test.js

// Track 3: a replica replicates the repo Hypercore DIRECTLY over Hyperswarm and reads the
// identical records locally — no HTTP, no RPC — and follows live appends. The "little heap".

import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { join } from 'node:path'
import { tmpdir } from 'node:os'
import b4a from 'b4a'
import Hyperswarm from 'hyperswarm'
import createTestnet from '@hyperswarm/testnet'
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 { serveReplication, connectReplica } from '../lib/fleet-replicate.js'

const NOW = '2026-07-02T00:00:00.000Z'
const PRIV = new Uint8Array(32).fill(7)

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('replica replicates the repo Hypercore over Hyperswarm — identical records + live append', async (t) => {
  const testnet = await createTestnet(3)
  const homeDir = mkdtempSync(join(tmpdir(), 'rep-home-'))
  const replicaDir = mkdtempSync(join(tmpdir(), 'rep-repl-'))
  const homeStore = openCorestore(homeDir)
  const replicaStore = openCorestore(replicaDir)

  // home: a repo Hypercore with one record, written through RepoManager
  const homeRepo = await new RepoStore(homeStore, { name: 'atproto-repo' }).ready()
  let micros = 1_700_000_000_000_000n
  const mgr = await new RepoManager(homeRepo, { did: 'did:plc:x', privateKey: PRIV, clock: () => (micros += 1_000_000n) }).init()
  await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3a', record: { $type: 'app.bsky.feed.post', text: 'one', createdAt: NOW } })
  const repoKey = homeRepo.core.key

  const homeSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
  const replicaSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
  const home = serveReplication(homeStore, homeSwarm)
  await home.ready()
  await connectReplica(replicaStore, replicaSwarm, home.serverKey)

  // the replica opens the home's repo core BY KEY (over replication) and reads it locally
  const replicaRepo = await new RepoStore(replicaStore, { key: repoKey }).ready()

  t.teardown(async () => {
    home.close()
    await homeSwarm.destroy(); await replicaSwarm.destroy()
    await replicaRepo.close(); await homeRepo.close()
    await homeStore.close(); await replicaStore.close()
    await testnet.destroy()
    rmSync(homeDir, { recursive: true, force: true }); rmSync(replicaDir, { recursive: true, force: true })
  })

  t.ok(await waitFor(async () => (await replicaRepo.records()).length === 1), 'replica received the record over replication')
  const homeRecs = await homeRepo.records()
  const replRecs = await replicaRepo.records()
  t.is(b4a.toString(replRecs[0].bytes, 'hex'), b4a.toString(homeRecs[0].bytes, 'hex'), 'record bytes identical — content replicated P2P, no HTTP')

  // live append: home writes a second record; the replica follows it over the swarm
  await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3b', record: { $type: 'app.bsky.feed.post', text: 'two', createdAt: NOW } })
  t.ok(await waitFor(async () => (await replicaRepo.records()).length === 2), 'live append propagated to the replica')
})