OverviewFilesCommitsIssuesDocsReleasesPeers

atproto-did-peer commits / 6a3ff36

did-peer: commitOps — decode a #commit frame into normalized ops with the record inline
DidPeer.commitOps(body) turns a decoded #commit frame body into [{action, collection, rkey, record?}], reading the changed record from the frame's OWN CAR (blocks) keyed by op.cid — race-free (no dependency on the repo core having replicated). This keeps the atproto CAR codec in the P2P-read layer so consumers (favela) ingest plain ops with no atproto-repo dependency. test/commitops.test.mjs: builds a real create+delete commit via RepoManager -> commitFrame, decodes it, and asserts commitOps recovers the record from the CAR and normalizes create/delete. Offline. Green 6/6. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
6a3ff3642d617c92c81ad7dedc73abc158a45b5e
by plankatron <fleeky@gmail.com> · 2026-07-14T15:03:13.000Z

Changed files (2) +71 -1

lib/did-peer.js +28 -1
……@@ -23,7 +23,7 @@ import { RepoStore } from 'atproto-store/lib/repo-store.js'
2323import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
2424import { decodeCommitFrame } from 'atproto-store/lib/firehose.js'
2525import { getRepoFromCommit, getRecord as rvGetRecord, listRecords as rvListRecords } from 'atproto-store/lib/read-view.js'
26import { readCar, loadCommit } from 'atproto-repo/lib/repo.js'
26import { readCar, loadCommit, decodeBlock } from 'atproto-repo/lib/repo.js'
2727import { verifyCommit, commitCid } from 'atproto-repo/lib/commit.js'
2828
2929const ANNOUNCE_PROTOCOL = 'atproto/did-announce/v1'
……@@ -183,6 +183,33 @@ export class DidPeer {
183183 return this.seq.subscribe(cursor, opts)
184184 }
185185
186 // Decode a #commit frame BODY (from firehose.decodeFrame) into normalized ops with the changed record
187 // INLINE — read from the frame's OWN CAR (body.blocks) keyed by op.cid, so it is race-free (does NOT
188 // depend on the repo core having caught up). Returns [{ action, collection, rkey, record? }]; delete
189 // ops omit record; create/update carry the decoded record (null if the block was absent). This keeps
190 // the atproto CAR codec in the P2P-read layer — consumers (favela) get plain ops, no atproto-repo dep.
191 static async commitOps (body) {
192 if (!body || !Array.isArray(body.ops)) return []
193 const needsCar = body.ops.some(o => (o.action === 'create' || o.action === 'update') && o.cid)
194 let car = null
195 if (needsCar && body.blocks) { try { car = await readCar(body.blocks) } catch { car = null } }
196 const out = []
197 for (const op of body.ops) {
198 const path = op.path || ''
199 const i = path.indexOf('/')
200 if (i <= 0) continue
201 const collection = path.slice(0, i)
202 const rkey = path.slice(i + 1)
203 if (op.action === 'delete') { out.push({ action: 'delete', collection, rkey }); continue }
204 if (op.action === 'create' || op.action === 'update') {
205 let record = null
206 if (op.cid && car) { try { record = decodeBlock(car, op.cid) } catch { record = null } }
207 out.push({ action: op.action, collection, rkey, record })
208 }
209 }
210 return out
211 }
212
186213 // Distinct collection NSIDs present in the replicated repo (keys are "<collection>/<rkey>").
187214 async collections () {
188215 if (!this.repo) return []
test/commitops.test.mjs +43 -0
……@@ -0,0 +1,43 @@
1// Offline unit test for DidPeer.commitOps — the race-free CAR extraction. Build a REAL #commit frame
2// (RepoManager.applyWrite -> commitFrame), decode it (firehose.decodeFrame), and assert commitOps
3// yields normalized ops with the changed record decoded FROM THE FRAME'S OWN CAR. Run:
4// node test/commitops.test.mjs
5import { mkdtempSync, rmSync } from 'node:fs'
6import { join } from 'node:path'
7import { tmpdir } from 'node:os'
8import { RepoStore } from 'atproto-store/lib/repo-store.js'
9import { RepoManager } from 'atproto-store/lib/repo-manager.js'
10import { commitFrame, decodeFrame } from 'atproto-store/lib/firehose.js'
11import { DidPeer } from '../lib/did-peer.js'
12
13const assert = (c, m) => { if (!c) { console.error('FAIL:', m); process.exit(1) } console.log(' ok -', m) }
14
15const DID = 'did:plc:commitops'
16const PRIV = new Uint8Array(32).fill(9)
17const iso = '2026-07-01T00:00:00.000Z'
18
19const dir = mkdtempSync(join(tmpdir(), 'commitops-'))
20const store = await new RepoStore(dir).ready()
21let micros = 1_700_000_000_000_000n
22const mgr = new RepoManager(store, { did: DID, privateKey: PRIV, clock: () => (micros += 1_000_000n) })
23await mgr.init()
24
25// CREATE -> the record must be recoverable from the frame CAR alone (no repo/store access)
26const wc = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3zzz', record: { $type: 'app.bsky.feed.post', text: 'from-car', createdAt: iso } })
27const dc = decodeFrame(await commitFrame(wc, { seq: 1, time: iso }))
28assert(dc && dc.t === '#commit', 'create frame decodes as #commit')
29const opsC = await DidPeer.commitOps(dc.body)
30assert(opsC.length === 1, 'commitOps returns 1 op for a single create')
31assert(opsC[0].action === 'create' && opsC[0].collection === 'app.bsky.feed.post' && opsC[0].rkey === '3zzz', 'op action/collection/rkey parsed from path')
32assert(opsC[0].record && opsC[0].record.text === 'from-car', 'RECORD decoded from the frame CAR (race-free)')
33
34// DELETE -> a delete op carries no record
35const wd = await mgr.applyWrite({ action: 'delete', collection: 'app.bsky.feed.post', rkey: '3zzz' })
36const dd = decodeFrame(await commitFrame(wd, { seq: 2, time: iso }))
37const opsD = await DidPeer.commitOps(dd.body)
38assert(opsD.length === 1 && opsD[0].action === 'delete' && opsD[0].rkey === '3zzz', 'commitOps yields a delete op')
39assert(opsD[0].record === undefined, 'delete op carries no record')
40
41console.log('\nCOMMITOPS OK: record extracted from the #commit CAR; create/delete normalized')
42await store.close()
43rmSync(dir, { recursive: true, force: true })