// One-time firehose repair. A swarm's __commit__ log can accumulate MALFORMED frames (full-set
// re-authors from before the restore fix, which a relay rejects on prevData). resetFirehose clears
// the log and re-emits a SINGLE clean continuation from a known-good baseline (the migration head +
// its records), so the relay resumes from the last commit it validly holds. Deterministic.
import '../lib/bare-globals.js'
import test from 'brittle'
import crypto from 'hypercore-crypto'
import fs from 'bare-fs'
import b4a from 'b4a'
import { StateBaseV3 } from 'atproto-room/lib/state-base.js'
import { SwarmDevice } from '../lib/swarm-device.js'
import { makeAttest, makeVerifyAttest, attestDeviceChain, deviceKeetKeyPair, rootIdentityPublicKey } from 'atproto-room/lib/membership.js'
import { deriveAtprotoKey } from 'atproto-identity'
const SIGN_SRC = 'abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about'
const KEET_ROOT = 'legal winner thank year wave sausage worth useful legal winner thank yellow'
const DEV = 'zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo wrong'
const DID = 'did:plc:resetfirehose2222222222'
const POST = 'app.bsky.feed.post'
const signKey = deriveAtprotoKey(SIGN_SRC)
const tmp = () => { const d = '/tmp/hpp-resetfh-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d }
const rec = (text) => ({ $type: POST, text, createdAt: '2026-07-05T00:00:00.000Z' })
test('resetFirehose clears the log + re-emits ONE clean continuation from the baseline', async (t) => {
const rootPub = await rootIdentityPublicKey(KEET_ROOT)
const dev = await deviceKeetKeyPair(DEV); const proof = await attestDeviceChain({ rootMnemonic: KEET_ROOT, devicePublicKey: dev.publicKey })
const members = [{ key: b4a.toString(dev.publicKey, 'hex'), role: 1 }]
const A = new StateBaseV3(tmp(), crypto.hash(b4a.from(DID)), {
roomSeed: crypto.hash(b4a.from('rf-seed')), encryptionKey: crypto.hash(b4a.from('rf-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 })
t.teardown(async () => { try { await A.close() } catch {} })
await device.claimLease()
// baseline: two "adopted" records, signed once — this is the known-good head the relay validly holds
await device.putRecord(POST, '3base00000001', rec('base a'))
await device.putRecord(POST, '3base00000002', rec('base b'))
const baseHead = await device.sign()
// accumulate more frames (two posts over two signs) — the "history" we want to collapse/repair
await device.putRecord(POST, '3post00000001', rec('post one')); await device.sign()
await device.putRecord(POST, '3post00000002', rec('post two')); await device.sign()
t.ok((await device.firehoseLog()).length >= 3, 'multiple #commit frames accumulated')
// repair: clear + re-emit ONE clean frame from the baseline (2 base records @ baseHead)
const h = await device.resetFirehose([
{ collection: POST, rkey: '3base00000001', record: rec('base a') },
{ collection: POST, rkey: '3base00000002', record: rec('base b') }
], baseHead)
t.ok(h, 'resetFirehose returned a clean head')
const after = await device.firehoseLog()
t.is(after.length, 1, 'exactly ONE #commit frame after the reset')
t.is(after[0].ops.length, 2, 'the single frame is an INCREMENTAL diff (the 2 posts since the baseline), not the full 4-record set')
t.alike(after[0].ops.map((o) => o.path).sort(), [POST + '/3post00000001', POST + '/3post00000002'], 'the clean frame carries exactly the two new posts')
t.is(after[0].prev, baseHead.commitCid.toString(), 'the clean frame continues from the baseline commit (prev = baseline head)')
})