// OFFLINE PROOF (node + testnet) — firehose-ingest convergence for the atproto virtual device.
//
// The ARCHITECTURE: hyper-overlay is pure content-addressed storage (records as deduped files); it
// does NOT converge trees. Convergence is the app layer = the signed atproto FIREHOSE. A SIGNER
// (RepoManager) authors a signed repo + sequences its firehose and serves the cores. A follower
// reads it KEYLESSLY (DidPeer) and RepoMirror writes the record BYTES into its own OverlayRepoStore
// files — transient (holds no firehose; the files ARE the converged state).
//
// Proves: (1) initial repo converges to record FILES; (2) a NEW write AFTER connect converges live
// via onHead (event-driven, the prod symptom the overlay bolt-on failed); (3) a delete propagates.
//
// Runs under NODE (the hyperswarm testnet DHT works under node; it did not under the pgh bare here).
// Run: node test/repo-mirror-ingest.test.mjs
import { mkdtempSync, rmSync } from 'node:fs'
import { join } from 'node:path'
import { tmpdir } from 'node:os'
import createTestnet from '@hyperswarm/testnet'
import Hyperswarm from 'hyperswarm'
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 { OverlayRepoStore } from 'atproto-store/lib/overlay-repo-store.js'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { createAtprotoIdentityFromKey } from 'atproto-identity/lib/atproto-identity.js'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import { DidPeer } from 'atproto-did-peer'
import { RepoMirror } from '../lib/repo-mirror.js'
const dirs = []
const toClose = []
let failed = 0
const newDir = (p) => { const d = mkdtempSync(join(tmpdir(), p)); dirs.push(d); return d }
const cleanup = () => { for (const c of toClose) { try { c() } catch {} } for (const d of dirs) { try { rmSync(d, { recursive: true, force: true }) } catch {} } }
const ok = (c, m) => { if (c) { console.log(' ok -', m) } else { console.error(' FAIL -', m); failed++ } }
const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
async function waitFor (fn, { tries = 60, gap = 500 } = {}) { for (let i = 0; i < tries; i++) { if (await fn()) return true; await sleep(gap) } return false }
const iso = '2026-07-19T00:00:00.000Z'
const id = createAtprotoIdentityFromKey(new Uint8Array(32).fill(7), { handle: 'signer.test', pdsUrl: 'https://pds.local' })
const testnet = await createTestnet(3)
const bootstrap = testnet.bootstrap
// ---- SIGNER: authors a signed repo, sequences its firehose, serves the cores ----
const sStore = openCorestore(newDir('signer-'))
const repoStore = await new RepoStore(sStore, { name: `atproto-repo:${id.did}` }).ready()
let micros = 1_700_000_000_000_000n
const mgr = await new RepoManager(repoStore, { did: id.did, privateKey: id.privateKey, clock: () => (micros += 1_000_000n) }).init()
const seq = await new FirehoseSequencer(sStore.get({ name: 'firehose' })).ready()
async function authorCreate (collection, rkey, record) { const w = await mgr.applyWrite({ action: 'create', collection, rkey, record }); await seq.sequenceCommit(w, { time: iso }) }
async function authorDelete (collection, rkey) { const w = await mgr.applyWrite({ action: 'delete', collection, rkey }); await seq.sequenceCommit(w, { time: iso }) }
await authorCreate('app.bsky.feed.post', '1', { $type: 'app.bsky.feed.post', text: 'first', createdAt: iso })
await authorCreate('app.bsky.feed.post', '2', { $type: 'app.bsky.feed.post', text: 'second', createdAt: iso })
await authorCreate('app.bsky.graph.follow', 'f1', { $type: 'app.bsky.graph.follow', subject: id.did, createdAt: iso })
const signerPeer = new DidPeer(sStore, { did: id.did, verifyKey: id.publicKey })
await signerPeer.provide({ repoKey: repoStore.core.key, firehoseKey: seq.core.key })
const sSwarm = new Hyperswarm({ bootstrap })
sSwarm.on('connection', (c) => signerPeer.attach(c))
await sSwarm.join(discoveryTopic(id.did), { server: true, client: true }).flushed()
toClose.push(() => sSwarm.destroy(), () => signerPeer.close(), () => sStore.close())
console.log('signer serving', id.did.slice(0, 24), 'with 3 records')
// ---- INGESTER: keyless DidPeer reader + RepoMirror -> OverlayRepoStore files (transient) ----
const iRoot = newDir('ingest-files-')
const fileStore = await new OverlayRepoStore(iRoot).ready() // records as FILES; no overlay needed
const iStore = openCorestore(newDir('ingest-store-'))
const reader = new DidPeer(iStore, { did: id.did, verifyKey: id.publicKey })
const iSwarm = new Hyperswarm({ bootstrap })
iSwarm.on('connection', (c) => reader.attach(c))
await iSwarm.join(discoveryTopic(id.did), { server: true, client: true }).flushed()
toClose.push(() => iSwarm.destroy(), () => reader.close(), () => iStore.close(), () => fileStore.close())
const head = await reader.ready({ timeout: 20000 })
ok(!!head, 'ingester reached a verified head (firehose/repo replicated)')
const mirror = new RepoMirror(reader, fileStore, { onSync: (s) => console.log(' mirror sync:', JSON.stringify(s)) })
await mirror.start()
toClose.push(() => mirror.close())
// helpers to read the ingester's FILE store
const has = async (coll, rkey) => !!(await fileStore.get(coll, rkey))
const textOf = async (coll, rkey) => { const r = await fileStore.get(coll, rkey); return r && r.text }
const fileCount = async () => (await fileStore.records()).length
// (1) INITIAL convergence
ok(await waitFor(async () => (await fileCount()) === 3), 'initial: all 3 records converged to FILES')
ok(await textOf('app.bsky.feed.post', '1') === 'first', "record post/1 body converged ('first')")
ok(await textOf('app.bsky.feed.post', '2') === 'second', "record post/2 body converged ('second')")
ok(await has('app.bsky.graph.follow', 'f1'), 'record follow/f1 converged')
// (2) NEW write AFTER connection — the prod symptom. onHead must drive a live re-sync.
await authorCreate('app.bsky.feed.post', '3', { $type: 'app.bsky.feed.post', text: 'live', createdAt: iso })
ok(await waitFor(async () => await has('app.bsky.feed.post', '3')), 'post-connection NEW write converged live (via onHead)')
ok(await textOf('app.bsky.feed.post', '3') === 'live', "new record post/3 body converged ('live')")
// (3) DELETE propagates
await authorDelete('app.bsky.feed.post', '1')
ok(await waitFor(async () => !(await has('app.bsky.feed.post', '1'))), 'delete propagated — post/1 removed from files')
ok(await fileCount() === 3, 'final file count is 3 (2 original + 1 live - 1 deleted)')
console.log(failed ? `\nFIREHOSE-INGEST: ${failed} FAILED` : '\nFIREHOSE-INGEST OK: repo converges to record files via the signed firehose (initial + live + delete)')
cleanup()
await testnet.destroy()
process.exit(failed ? 1 : 0)