48f5fc4 · atproto-base

Changed files (7) +215 -0

lib/base.js +32 -0

…	…	@@ -21,7 +21,10 @@ import Autobase from 'autobase'
21	21	import Hyperbee from 'hyperbee'
22	22	import b4a from 'b4a'
23	23	import crypto from 'hypercore-crypto'
	24	import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
24	25	import { projectCommit, nextRev } from './project.js'
	26	import { buildCommitWrite } from './firehose.js'
	27	import { revTime } from './rev.js'
25	28
26	29	// Typed ops. start-indexing/stop-indexing mirror keet's members/index.js handlers I/P EXACTLY: a
27	30	// device SELF-promotes/demotes its own writer, gated on its member being an admin (role 1 = keet
…	…	@@ -57,6 +60,8 @@ export class AtprotoBase {
57	60	this._members = opts.members \|\| null
58	61	this._ackInterval = opts.ackInterval \|\| 1000
59	62	this._admitted = new Set()
	63	this._firehose = null // FirehoseSequencer (Hypercore of #commit frames)
	64	this._prevHead = { records: [], rev: null, commitCid: null } // last EMITTED firehose head (for diffing)
60	65	this.store = null
61	66	this.base = null
62	67	this._ready = false
…	…	@@ -79,6 +84,9 @@ export class AtprotoBase {
79	84	if (this._encryptionKey) { baseOpts.encryptionKey = this._encryptionKey; baseOpts.encrypted = true }
80	85	this.base = new Autobase(this.store, this._bootstrap, baseOpts)
81	86	await this.base.ready()
	87	// The public firehose: a plain Hypercore of #commit/#sync frames the bridge replicates + serves as
	88	// com.atproto.sync.subscribeRepos. Separate from the autobase (a projection of it), like the membrane.
	89	this._firehose = await new FirehoseSequencer(this.store.get({ name: 'atproto-firehose' })).ready()
82	90	this._ready = true
83	91	if (this._members) this.projectMembers(this._members).catch(() => {}) // creator best-effort projection
84	92	}
…	…	@@ -345,7 +353,31 @@ export class AtprotoBase {
345	353	return projectCommit({ records, did: this._did, rev, prev, privateKey: this._privateKey })
346	354	}
347	355
	356	// ── Firehose (com.atproto.sync.subscribeRepos) ──────────────────────────────────────────────────
	357	// Emit a #commit frame for the current FINALIZED head if it advanced since the last emit. Deterministic
	358	// (buildCommitWrite is pure; the seq/time the sequencer stamps are firehose-server state). Returns the
	359	// write, or null if the finalized set is unchanged. Call after finalization (the chain advances on
	360	// finalization, not on write) — e.g. wire it to base 'update' in production.
	361	async emitFirehose () {
	362	await this.settled()
	363	const records = await this.indexedRecords()
	364	const rev = nextRev(this._prevHead.rev)
	365	const write = await buildCommitWrite({
	366	oldRecords: this._prevHead.records, newRecords: records,
	367	did: this._did, rev, since: this._prevHead.rev, prev: this._prevHead.commitCid, privateKey: this._privateKey
	368	})
	369	if (write.ops.length === 0 && this._prevHead.commitCid) return null // nothing finalized-new to emit
	370	await this._firehose.sequenceCommit(write, { time: revTime(rev) })
	371	this._prevHead = { records, rev, commitCid: write.commitCid }
	372	return write
	373	}
	374
	375	subscribeRepos (cursor = 0, opts) { return this._firehose.subscribe(cursor, opts) }
	376	get firehoseKey () { return this._firehose ? this._firehose.core.key : null }
	377	get firehoseLength () { return this._firehose ? this._firehose.length : 0 }
	378
348	379	async close () {
	380	if (this._firehose) { try { await this._firehose.close() } catch {} this._firehose = null }
349	381	if (this.base) { try { await this.base.close() } catch {} this.base = null }
350	382	if (this.store && !this._injectedStore) { try { await this.store.close() } catch {} }
351	383	this._ready = false

lib/firehose.js +72 -0

…	…	@@ -0,0 +1,72 @@
	1	// The FIREHOSE projection — turn consecutive finalized heads into atproto `#commit` frames
	2	// (com.atproto.sync.subscribeRepos) that the bridge/relay replicate. Deterministic like the commit
	3	// projection: a #commit frame is a pure function of (oldRecords → newRecords), so co-owners produce
	4	// identical frames (the seq/time the sequencer stamps are firehose-server state, not signed).
	5	//
	6	// A #commit carries the DIFF, not the full repo: `ops` (create/update/delete), the MINIMAL MST proof
	7	// blocks (atproto-repo firehoseProofBlocks — created + delete-exclusion + inductive-inversion nodes)
	8	// PLUS the changed record blocks, and `prevData` (old MST root) for inductive verification. We reuse
	9	// atproto-store's byte-exact commitFrame/decodeCommitFrame codec (what the bridge already consumes).
	10
	11	import { cidForValue } from 'atproto-repo/lib/cid.js'
	12	import { firehoseProofBlocks } from 'atproto-repo/lib/mst.js'
	13	import { unsignedCommit, signCommit, commitCid } from 'atproto-repo/lib/commit.js'
	14
	15	// records: [{ key, value: recordObject }] → [{ key, cid, value }]
	16	async function withCids (records) {
	17	const out = []
	18	for (const r of records) { const { cid, bytes } = await cidForValue(r.value); out.push({ key: r.key, cid, bytes, value: r.value }) }
	19	return out
	20	}
	21
	22	// The atproto ops diff between two record sets (by MST key).
	23	function diffOps (oldW, newW) {
	24	const oldMap = new Map(oldW.map((r) => [r.key, r.cid]))
	25	const newMap = new Map(newW.map((r) => [r.key, r.cid]))
	26	const ops = []
	27	for (const { key, cid } of newW) {
	28	const prev = oldMap.get(key)
	29	if (!prev) ops.push({ action: 'create', path: key, cid, prev: null })
	30	else if (!prev.equals(cid)) ops.push({ action: 'update', path: key, cid, prev })
	31	}
	32	for (const { key, cid } of oldW) { if (!newMap.has(key)) ops.push({ action: 'delete', path: key, cid: null, prev: cid }) }
	33	return ops
	34	}
	35
	36	// Build the `write` object atproto-store's commitFrame() wants, for the transition oldRecords→newRecords.
	37	// Deterministic. `prev` = previous commit CID (null = genesis), `since` = previous rev (relay gap hint).
	38	export async function buildCommitWrite ({ oldRecords = [], newRecords, did, rev, since = null, prev = null, privateKey }) {
	39	const oldW = await withCids(oldRecords)
	40	const newW = await withCids(newRecords)
	41	const ops = diffOps(oldW, newW)
	42
	43	// Minimal MST proof blocks + the new/old roots (firehoseProofBlocks throws if the proof is invalid).
	44	const { blocks, dataCid, oldRoot } = firehoseProofBlocks({
	45	oldRecords: oldW.map((r) => ({ key: r.key, value: r.cid })),
	46	newRecords: newW.map((r) => ({ key: r.key, value: r.cid })),
	47	ops
	48	})
	49
	50	// The changed RECORD blocks must also travel (created/updated record content).
	51	const changed = new Map(blocks) // start from the MST proof block set
	52	for (const op of ops) {
	53	if (op.action === 'delete') continue
	54	const rec = newW.find((r) => r.key === op.path)
	55	changed.set(op.cid.toString(), { cid: op.cid, bytes: rec.bytes })
	56	}
	57
	58	const signed = signCommit(unsignedCommit({ did, rev, data: dataCid, prev }), privateKey)
	59	const cCid = commitCid(signed)
	60	const { bytes: commitBytes } = await cidForValue(signed)
	61
	62	return {
	63	commitCid: cCid,
	64	commitBlock: { cid: cCid, bytes: commitBytes },
	65	newBlocks: [...changed.values()],
	66	ops,
	67	rev,
	68	since,
	69	signedCommit: signed,
	70	prevData: oldRoot
	71	}
	72	}

lib/rev.js +6 -0

…	…	@@ -25,3 +25,9 @@ export function nextRev (prevRev, baseMicros = 0) {
25	25	const micros = prevRev ? decodeTidMicros(prevRev) : baseMicros
26	26	return encodeTid(prevRev ? micros + 1 : micros)
27	27	}
	28
	29	// A DETERMINISTIC ISO timestamp derived from a rev — used for the firehose frame `time` (cosmetic,
	30	// not signed), so co-owners emit byte-identical frames. NOT wall-clock.
	31	export function revTime (rev) {
	32	return new Date(Math.floor(decodeTidMicros(rev) / 1000)).toISOString()
	33	}

package-lock.json +21 -0

…	…	@@ -12,6 +12,7 @@
12	12	"@noble/hashes": "1.8.0",
13	13	"atproto-identity": "file:../atproto-identity",
14	14	"atproto-repo": "file:../atproto-repo",
	15	"atproto-store": "file:../atproto-store",
15	16	"hypercore-crypto": "^3.6.0"
16	17	},
17	18	"devDependencies": {
…	…	@@ -54,6 +55,22 @@
54	55	"brittle": "^4.0.2"
55	56	}
56	57	},
	58	"../atproto-store": {
	59	"version": "0.0.8",
	60	"dependencies": {
	61	"@ipld/dag-cbor": "^9.2.1",
	62	"atproto-repo": "file:../atproto-repo",
	63	"b4a": "^1.6.0",
	64	"bare-fs": "^4.7.3",
	65	"bare-path": "^3.0.1",
	66	"corestore": "^7.4.0",
	67	"hyperbee": "^2.20.0",
	68	"random-access-memory": "^6.2.0"
	69	},
	70	"devDependencies": {
	71	"brittle": "^4.0.2"
	72	}
	73	},
57	74	"node_modules/@hyperswarm/secret-stream": {
58	75	"version": "6.9.1",
59	76	"resolved": "https://registry.npmjs.org/@hyperswarm/secret-stream/-/secret-stream-6.9.1.tgz",
…	…	@@ -142,6 +159,10 @@
142	159	"resolved": "../atproto-repo",
143	160	"link": true
144	161	},
	162	"node_modules/atproto-store": {
	163	"resolved": "../atproto-store",
	164	"link": true
	165	},
145	166	"node_modules/autobase": {
146	167	"version": "7.28.1",
147	168	"resolved": "https://registry.npmjs.org/autobase/-/autobase-7.28.1.tgz",

package.json +1 -0

…	…	@@ -10,6 +10,7 @@
10	10	"dependencies": {
11	11	"atproto-repo": "file:../atproto-repo",
12	12	"atproto-identity": "file:../atproto-identity",
	13	"atproto-store": "file:../atproto-store",
13	14	"@noble/curves": "1.9.7",
14	15	"@noble/hashes": "1.8.0",
15	16	"hypercore-crypto": "^3.6.0"

test/base.test.mjs +22 -0

…	…	@@ -10,6 +10,7 @@ import { deriveAtprotoKey, createAtprotoIdentity } from 'atproto-identity'
10	10	import { verifyCommit } from 'atproto-repo/lib/commit.js'
11	11	import { readCar } from 'atproto-repo/lib/car.js'
12	12	import { verifyRepo } from 'atproto-repo'
	13	import { decodeCommitFrame } from 'atproto-store/lib/firehose.js'
13	14	import { AtprotoBase } from '../lib/base.js'
14	15
15	16	const MNEMONIC = 'test test test test test test test test test test test junk'
…	…	@@ -66,3 +67,24 @@ test('AtprotoBase: two independent bases with the same records project the IDENT
66	67	t.is(ha.commitCid.toString(), hb.commitCid.toString(), 'identical head commit CID across co-owners')
67	68	t.is(Buffer.compare(Buffer.from(ha.signedCommit.sig), Buffer.from(hb.signedCommit.sig)), 0, 'identical signature (no signer election needed)')
68	69	})
	70
	71	test('AtprotoBase firehose: finalized head advances emit #commit frames; subscribeRepos replays them', async (t) => {
	72	const A = mkBase(); await A.base.ready()
	73	t.teardown(() => A.close())
	74
	75	await A.base.putRecord('app.bsky.feed.post', '3a', post('one'))
	76	const w1 = await A.base.emitFirehose()
	77	t.ok(w1 && w1.ops.length === 1, 'first emit: one create op for the finalized head')
	78
	79	await A.base.putRecord('app.bsky.feed.post', '3b', post('two'))
	80	const w2 = await A.base.emitFirehose()
	81	t.ok(w2 && w2.ops.some((o) => o.action === 'create' && o.path.endsWith('3b')), 'second emit: create 3b')
	82
	83	t.is(await A.base.emitFirehose(), null, 'no finalized change → no new frame emitted')
	84	t.is(A.base.firehoseLength, 2, 'firehose core has exactly 2 #commit frames')
	85
	86	const frames = []
	87	for await (const { seq, bytes } of A.base.subscribeRepos(0)) { frames.push(bytes); if (frames.length >= 2) break }
	88	t.is(frames.length, 2, 'subscribeRepos replayed both frames')
	89	t.is(decodeCommitFrame(frames[0]).body.commit.toString(), w1.commitCid.toString(), 'replayed frame 1 = the first finalized head')
	90	})

test/firehose.test.mjs +61 -0

…	…	@@ -0,0 +1,61 @@
	1	// FIREHOSE — consecutive finalized heads become atproto #commit frames (the DIFF: ops + minimal
	2	// proof blocks + prevData), byte-exact via atproto-store's codec (what the bridge consumes), and
	3	// DETERMINISTIC across co-owners. Covers create/update/delete and a genesis.
	4
	5	import test from 'brittle'
	6	import { deriveAtprotoKey, createAtprotoIdentity } from 'atproto-identity'
	7	import { readCar } from 'atproto-repo/lib/car.js'
	8	import { commitFrame, decodeCommitFrame } from 'atproto-store/lib/firehose.js'
	9	import { buildCommitWrite } from '../lib/firehose.js'
	10	import { nextRev } from '../lib/rev.js'
	11
	12	const MNEMONIC = 'test test test test test test test test test test test junk'
	13	const ID = createAtprotoIdentity(MNEMONIC, { handle: 'x.7rnx.net', pdsUrl: 'https://pds.7rnx.net' })
	14	const KEY = deriveAtprotoKey(MNEMONIC)
	15	const rec = (t) => ({ $type: 'app.bsky.feed.post', text: t, createdAt: '2026-07-23T00:00:00.000Z', langs: ['en'] })
	16	const R = (k, v) => ({ key: 'app.bsky.feed.post/' + k, value: rec(v) })
	17	const opFor = (w, path) => w.ops.find((o) => o.path === path)
	18
	19	test('firehose #commit: create/update/delete diffs, decodable frames, deterministic', async (t) => {
	20	const did = ID.did; const pk = KEY.privateKey
	21
	22	// genesis: create 3a
	23	const rev1 = nextRev(null)
	24	const w1 = await buildCommitWrite({ oldRecords: [], newRecords: [R('3a', 'one')], did, rev: rev1, prev: null, privateKey: pk })
	25	t.is(w1.ops.length, 1); t.is(opFor(w1, 'app.bsky.feed.post/3a').action, 'create')
	26
	27	// add 3b
	28	const rev2 = nextRev(rev1)
	29	const w2 = await buildCommitWrite({ oldRecords: [R('3a', 'one')], newRecords: [R('3a', 'one'), R('3b', 'two')], did, rev: rev2, since: rev1, prev: w1.commitCid, privateKey: pk })
	30	t.is(opFor(w2, 'app.bsky.feed.post/3b').action, 'create')
	31	t.absent(opFor(w2, 'app.bsky.feed.post/3a'), 'unchanged record produces no op')
	32
	33	// update 3a
	34	const rev3 = nextRev(rev2)
	35	const w3 = await buildCommitWrite({ oldRecords: [R('3a', 'one'), R('3b', 'two')], newRecords: [R('3a', 'one-edited'), R('3b', 'two')], did, rev: rev3, since: rev2, prev: w2.commitCid, privateKey: pk })
	36	const up = opFor(w3, 'app.bsky.feed.post/3a')
	37	t.is(up.action, 'update'); t.ok(up.prev, 'update carries prev CID for inductive proof')
	38
	39	// delete 3b
	40	const rev4 = nextRev(rev3)
	41	const w4 = await buildCommitWrite({ oldRecords: [R('3a', 'one-edited'), R('3b', 'two')], newRecords: [R('3a', 'one-edited')], did, rev: rev4, since: rev3, prev: w3.commitCid, privateKey: pk })
	42	const del = opFor(w4, 'app.bsky.feed.post/3b')
	43	t.is(del.action, 'delete'); t.ok(del.prev, 'delete carries prev CID for exclusion proof')
	44
	45	// frame + decode round-trip
	46	const frame = await commitFrame(w2, { seq: 2, time: '2026-07-23T00:00:00.000Z' })
	47	const dec = decodeCommitFrame(frame)
	48	t.is(dec.header.t, '#commit', 'decodes as #commit')
	49	t.is(dec.body.commit.toString(), w2.commitCid.toString(), 'frame commit CID matches')
	50	t.is(dec.body.rev, rev2, 'frame rev matches')
	51	t.is(dec.body.repo, did, 'frame repo = did')
	52	t.is(dec.body.ops.length, 1, 'frame carries the diff ops')
	53	const car = await readCar(dec.body.blocks)
	54	t.is(car.roots[0].toString(), w2.commitCid.toString(), 'frame CAR root = the commit')
	55	t.ok(car.blocks.has(opFor(w2, 'app.bsky.feed.post/3b').cid.toString()), 'the new record block travels in the frame')
	56
	57	// DETERMINISM: a co-owner rebuilding the same transition gets the identical commit + ops.
	58	const w2b = await buildCommitWrite({ oldRecords: [R('3a', 'one')], newRecords: [R('3a', 'one'), R('3b', 'two')], did, rev: rev2, since: rev1, prev: w1.commitCid, privateKey: pk })
	59	t.is(w2.commitCid.toString(), w2b.commitCid.toString(), 'firehose commit deterministic across co-owners')
	60	t.is(Buffer.compare(Buffer.from(w2.signedCommit.sig), Buffer.from(w2b.signedCommit.sig)), 0, 'identical signature')
	61	})

atproto-base commits / 48f5fc4

Changed files (7) +215 -0