| … | … | @@ -23,6 +23,8 @@ import Hypercore from 'hypercore' |
| 23 | 23 | import b4a from 'b4a' |
| 24 | 24 | import crypto from 'hypercore-crypto' |
| 25 | 25 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' |
| 26 | import { getRepoFromCommit } from 'atproto-store/lib/read-view.js' |
| 27 | import { cidForValue } from 'atproto-repo/lib/cid.js' |
| 26 | 28 | import { projectCommit, nextRev } from './project.js' |
| 27 | 29 | import { buildCommitWrite } from './firehose.js' |
| 28 | 30 | import { revTime } from './rev.js' |
| … | … | @@ -57,6 +59,20 @@ function deriveBaseKey (gsPublicKey) { return Hypercore.key({ version: 1, signer |
| 57 | 59 | // key — a non-member has neither, so they can neither locate nor decrypt the co-owned base. |
| 58 | 60 | function deriveEncKey (roomSeed) { return crypto.hash([ENC_NAMESPACE, toBuf(roomSeed)]) } |
| 59 | 61 | |
| 62 | // A minimal RepoStore-compatible projection (put/del/records) over atproto-base's OWN corestore. We |
| 63 | // don't reuse atproto-store's RepoStore because its `storage instanceof Corestore` guard fails across |
| 64 | // duplicated corestore module copies (atproto-store is a symlinked file: dep with its own node_modules). |
| 65 | // getRepoFromCommit only reads records() -> [{key, bytes}]; record bytes are canonical DAG-CBOR |
| 66 | // (cidForValue), so CIDs match atproto-repo's regardless of which module copy computed them. |
| 67 | class RepoProjection { |
| 68 | constructor (core) { this.core = core; this.bee = new Hyperbee(core, { keyEncoding: 'utf-8', valueEncoding: 'binary', alwaysDuplicate: false }) } |
| 69 | async ready () { await this.core.ready(); await this.bee.ready(); return this } |
| 70 | async put (collection, rkey, record) { const { bytes } = await cidForValue(record); await this.bee.put(`${collection}/${rkey}`, b4a.from(bytes)) } |
| 71 | async del (collection, rkey) { await this.bee.del(`${collection}/${rkey}`) } |
| 72 | async records () { const out = []; for await (const n of this.bee.createReadStream()) out.push({ key: String(n.key), bytes: n.value }); return out } |
| 73 | async close () { try { await this.bee.close() } catch {} } |
| 74 | } |
| 75 | |
| 60 | 76 | export class AtprotoBase { |
| 61 | 77 | // opts: { path | store, bootstrap?, keyPair?, encryptionKey?, did, privateKey, identity?, members? } |
| 62 | 78 | // identity: { publicKey, secretKey } — this co-owner's membership key (ed25519, drives admission). |
| … | … | @@ -91,7 +107,8 @@ export class AtprotoBase { |
| 91 | 107 | this._ackInterval = opts.ackInterval || 1000 |
| 92 | 108 | this._admitted = new Set() |
| 93 | 109 | this._firehose = null // FirehoseSequencer (Hypercore of #commit frames) |
| 94 | | this._prevHead = { records: [], rev: null, commitCid: null } // last EMITTED firehose head (for diffing) |
| 110 | this._repo = null // RepoStore projection (getRepo for the bridge) |
| 111 | this._prevHead = { records: [], rev: null, commitCid: null, signedCommit: null } // last EMITTED head |
| 95 | 112 | this.store = null |
| 96 | 113 | this.base = null |
| 97 | 114 | this._ready = false |
| … | … | @@ -127,6 +144,9 @@ export class AtprotoBase { |
| 127 | 144 | // The public firehose: a plain Hypercore of #commit/#sync frames the bridge replicates + serves as |
| 128 | 145 | // com.atproto.sync.subscribeRepos. Separate from the autobase (a projection of it), like the membrane. |
| 129 | 146 | this._firehose = await new FirehoseSequencer(this.store.get({ name: 'atproto-firehose' })).ready() |
| 147 | // The public repo core the bridge's getRepo assembles from (a RepoStore kept in sync with the |
| 148 | // finalized record set on each emitFirehose, so its MST root matches the emitted signed commit). |
| 149 | this._repo = await new RepoProjection(this.store.get({ name: 'atproto-repo-projection' })).ready() |
| 130 | 150 | this._ready = true |
| 131 | 151 | if (this._members) this.projectMembers(this._members).catch(() => {}) // creator best-effort projection |
| 132 | 152 | } |
| … | … | @@ -431,17 +451,35 @@ export class AtprotoBase { |
| 431 | 451 | did: this._did, rev, since: this._prevHead.rev, prev: this._prevHead.commitCid, privateKey: this._privateKey |
| 432 | 452 | }) |
| 433 | 453 | if (write.ops.length === 0 && this._prevHead.commitCid) return null // nothing finalized-new to emit |
| 454 | await this._projectRepo(records) // keep the repo core in sync FIRST so getRepo() matches this commit |
| 434 | 455 | await this._firehose.sequenceCommit(write, { time: revTime(rev) }) |
| 435 | | this._prevHead = { records, rev, commitCid: write.commitCid } |
| 456 | this._prevHead = { records, rev, commitCid: write.commitCid, signedCommit: write.signedCommit } |
| 436 | 457 | return write |
| 437 | 458 | } |
| 438 | 459 | |
| 460 | // Sync the RepoStore projection to the finalized record set (put changed, delete removed) so its MST |
| 461 | // root equals the emitted commit's data CID (getRepoFromCommit asserts this). |
| 462 | async _projectRepo (records) { |
| 463 | const want = new Set(records.map((r) => r.key)) |
| 464 | for (const r of records) { const i = r.key.lastIndexOf('/'); await this._repo.put(r.key.slice(0, i), r.key.slice(i + 1), r.value) } |
| 465 | for (const e of await this._repo.records()) { if (!want.has(e.key)) { const i = e.key.lastIndexOf('/'); await this._repo.del(e.key.slice(0, i), e.key.slice(i + 1)) } } |
| 466 | } |
| 467 | |
| 468 | // com.atproto.sync.getRepo — the signed CAR the bridge serves, assembled from the RepoStore + the last |
| 469 | // emitted signed commit (exactly PeerFront.getRepo's contract). Call emitFirehose() at least once first. |
| 470 | async getRepo () { |
| 471 | if (!this._prevHead.signedCommit) throw new Error('AtprotoBase.getRepo: no commit emitted yet — call emitFirehose() first') |
| 472 | return getRepoFromCommit(this._repo, this._prevHead.signedCommit) |
| 473 | } |
| 474 | |
| 439 | 475 | subscribeRepos (cursor = 0, opts) { return this._firehose.subscribe(cursor, opts) } |
| 440 | 476 | get firehoseKey () { return this._firehose ? this._firehose.core.key : null } |
| 441 | 477 | get firehoseLength () { return this._firehose ? this._firehose.length : 0 } |
| 478 | get repoKey () { return this._repo ? this._repo.core.key : null } |
| 442 | 479 | |
| 443 | 480 | async close () { |
| 444 | 481 | if (this._firehose) { try { await this._firehose.close() } catch {} this._firehose = null } |
| 482 | if (this._repo) { try { await this._repo.close() } catch {} this._repo = null } |
| 445 | 483 | if (this.base) { try { await this.base.close() } catch {} this.base = null } |
| 446 | 484 | if (this.store && !this._injectedStore) { try { await this.store.close() } catch {} } |
| 447 | 485 | this._ready = false |