// Slice 3: write path (RepoManager) + #commit firehose frames. Verified structurally
// (header byte-exact, body field-set, blocks is a valid CAR, ops correct, head-state
// advances) and against a LIVE millipds #commit frame (field-set parity).
import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { join } from 'node:path'
import { tmpdir } from 'node:os'
import { RepoStore } from '../lib/repo-store.js'
import { RepoManager } from '../lib/repo-manager.js'
import { commitFrame, decodeCommitFrame, decodeFrame, syncFrame, restampFrame, COMMIT_HEADER, SYNC_HEADER } from '../lib/firehose.js'
import { isValidTid } from '../lib/tid.js'
import { readCar, loadCommit } from 'atproto-repo/lib/repo.js'
import { commitCid } from 'atproto-repo/lib/commit.js'
import { getRecord } from '../lib/read-view.js'
const DID = 'did:plc:yu4kgcktm7qtrwo5ofyiaf4c'
const PRIV = new Uint8Array(32).fill(7)
const EXPECTED_BODY_KEYS = ['blobs', 'blocks', 'commit', 'ops', 'prev', 'prevData', 'rebase', 'repo', 'rev', 'seq', 'since', 'time', 'tooBig'].sort()
async function newMgr (t) {
const dir = mkdtempSync(join(tmpdir(), 'firehose-'))
const store = await new RepoStore(dir).ready()
t.teardown(async () => { try { await store.close() } catch {} rmSync(dir, { recursive: true, force: true }) })
let micros = 1_700_000_000_000_000n // deterministic monotonic clock
const mgr = new RepoManager(store, { did: DID, privateKey: PRIV, clock: () => (micros += 1_000_000n) })
await mgr.init()
return { store, mgr }
}
const iso = '2026-07-01T00:00:00.000Z'
// 1. create -> frame header byte-exact + full body field set
test('create emits a #commit frame: header byte-exact, body fields complete', async (t) => {
const { mgr } = await newMgr(t)
const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3aaa', record: { $type: 'app.bsky.feed.post', text: 'hi', createdAt: iso } })
const frame = await commitFrame(w, { seq: 1, time: iso })
const headerOk = Buffer.from(frame.subarray(0, COMMIT_HEADER.length)).equals(Buffer.from(COMMIT_HEADER))
const dec = decodeCommitFrame(frame)
const keysOk = JSON.stringify(Object.keys(dec.body).sort()) === JSON.stringify(EXPECTED_BODY_KEYS)
t.ok(headerOk && dec !== null && keysOk, 'create emits a #commit frame: header byte-exact, body fields complete')
})
// 2. body values + ops
test('body: ops/rev/since/commit/repo correct; rev is a valid TID', async (t) => {
const { mgr } = await newMgr(t)
const before = mgr.rev
const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3bbb', record: { $type: 'app.bsky.feed.post', text: 'yo', createdAt: iso } })
const { body } = decodeCommitFrame(await commitFrame(w, { seq: 5, time: iso }))
t.ok(body.repo === DID &&
body.seq === 5 &&
body.rebase === false && body.tooBig === false && body.prev === null &&
body.since === before && isValidTid(body.rev) && body.rev > before &&
body.commit.equals(w.commitCid) &&
body.ops.length === 1 && body.ops[0].action === 'create' &&
body.ops[0].path === 'app.bsky.feed.post/3bbb' && body.ops[0].cid !== null,
'body: ops/rev/since/commit/repo correct; rev is a valid TID')
})
// 3. blocks is a valid CAR rooted at the commit, containing commit + record
test('body.blocks is a valid CAR rooted at the new commit', async (t) => {
const { mgr } = await newMgr(t)
const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3ccc', record: { $type: 'app.bsky.feed.post', text: 'z', createdAt: iso } })
const { body } = decodeCommitFrame(await commitFrame(w, { seq: 2, time: iso }))
const car = await readCar(body.blocks)
const commit = loadCommit(car)
t.ok(car.roots[0].equals(body.commit) && commit.value.did === DID && commit.value.data.equals(w.dataCid),
'body.blocks is a valid CAR rooted at the new commit')
})
// 4. delete op + monotonic rev + head-state reflected in reads
test('delete: op cid null, rev advances, read-view reflects removal', async (t) => {
const { store, mgr } = await newMgr(t)
await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3ddd', record: { $type: 'app.bsky.feed.post', text: 'tmp', createdAt: iso } })
const revAfterCreate = mgr.rev
const w = await mgr.applyWrite({ action: 'delete', collection: 'app.bsky.feed.post', rkey: '3ddd' })
const { body } = decodeCommitFrame(await commitFrame(w, { seq: 3, time: iso }))
const gone = await getRecord(store, 'app.bsky.feed.post', '3ddd', { did: DID })
t.ok(body.ops[0].action === 'delete' && body.ops[0].cid === null &&
body.rev > revAfterCreate && body.since === revAfterCreate && gone === null,
'delete: op cid null, rev advances, read-view reflects removal')
})
// 4b. #sync frame (Sync 1.1 re-baseline): header byte-exact, body fields, CAR carries the
// commit block (root = commit CID), rev matches, and restampFrame updates its seq.
test('#sync frame: header + body + CAR-with-commit + restamp', async (t) => {
const { mgr } = await newMgr(t)
const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3sss', record: { $type: 'app.bsky.feed.post', text: 's', createdAt: iso } })
const signed = w.signedCommit
const frame = await syncFrame({ seq: 7, did: DID, signedCommit: signed, time: iso })
const headerOk = Buffer.from(frame.subarray(0, SYNC_HEADER.length)).equals(Buffer.from(SYNC_HEADER))
const dec = decodeFrame(frame)
const car = await readCar(dec.body.blocks)
const commit = loadCommit(car)
const cid = commitCid(signed)
const restamped = decodeFrame(restampFrame(frame, 99))
t.ok(
headerOk && dec.t === '#sync' &&
dec.body.did === DID && dec.body.seq === 7 && dec.body.rev === signed.rev &&
car.roots[0].equals(cid) && commit.value.rev === signed.rev && commit.value.did === DID &&
restamped.body.seq === 99,
'#sync frame: header + body + CAR-with-commit + restamp')
})
// The live subscribeRepos stream leaves a refed TLSSocket that ws.close() doesn't
// release; brittle ends the run on `beforeExit` (loop drain), so unref it or the
// runner hangs after all asserts pass. (This is why the pre-brittle version force-exited.)
function unrefLingeringTlsSockets () {
if (typeof process._getActiveHandles !== 'function') return
for (const h of process._getActiveHandles()) {
if (h && h.constructor && h.constructor.name === 'TLSSocket') { try { h.unref() } catch {} }
}
}
// 5. LIVE millipds #commit frame: our body field-set matches (non-fatal skip)
test('LIVE millipds #commit body field-set matches ours (or skip)', async (t) => {
let liveKeys, unreachable = null
try {
liveKeys = await new Promise((resolve, reject) => {
const ws = new WebSocket('wss://pds.7rnx.net/xrpc/com.atproto.sync.subscribeRepos?cursor=0')
ws.binaryType = 'arraybuffer'
const to = setTimeout(() => { try { ws.close() } catch {} ; reject(new Error('timeout')) }, 9000)
ws.onmessage = (ev) => {
const dec = decodeCommitFrame(new Uint8Array(ev.data))
if (dec) { clearTimeout(to); try { ws.close() } catch {} ; resolve(Object.keys(dec.body).sort()) }
}
ws.onerror = (e) => { clearTimeout(to); reject(new Error('ws error')) }
})
} catch (e) {
unreachable = e.message
} finally {
unrefLingeringTlsSockets()
}
if (unreachable !== null) {
t.comment(`millipds firehose unreachable: ${unreachable} — skipped`)
t.pass('millipds firehose unreachable — skipped')
return
}
t.comment('live #commit keys: ' + liveKeys.join(','))
// our keys must be a superset-or-equal of millipds's core set (millipds may add fields)
const ours = new Set(EXPECTED_BODY_KEYS)
const missing = liveKeys.filter(k => !ours.has(k))
if (missing.length) t.comment('fields millipds has that we lack: ' + missing.join(','))
t.is(missing.length, 0, 'LIVE millipds #commit body field-set matches ours')
})