hyper-proto-peer / lib / repo-mirror.js

// RepoMirror — the atproto CONVERGENCE layer for a records-as-files device.
//
// hyper-overlay is a pure content-addressed STORAGE/distribution library; it does not converge
// trees (that's an application concern — see hyper-overlay v3-audit). For atproto the convergence
// channel is the signed FIREHOSE: the signer materializes a signed repo + firehose; a non-signer
// (or any follower) reads it keylessly (a DidPeer-shaped reader) and writes the record BYTES into
// its own OverlayRepoStore files. The reader's cores are TRANSIENT — this holds no firehose; the
// record files ARE the converged state, re-derivable into a byte-identical signed repo.
//
// Event-driven, never polled: it re-syncs on the reader's `onHead` (fires on each verified head,
// i.e. every replicated firehose commit). Duck-typed reader (no DidPeer import → no circular dep):
//   reader.head              -> truthy once a verified head exists
//   reader.collections()     -> [collectionNSID]
//   reader.listRecords(coll) -> [{ uri, cid, value }]  (rkey = last uri segment)
//   reader._onHead           -> settable (err)=>void trigger (DidPeer calls it on head advance)
//
// store = an OverlayRepoStore (records as deduped files on disk; overlay optional).

import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
import * as dagCbor from '@ipld/dag-cbor'

export class RepoMirror {
  constructor (reader, store, { onSync = null } = {}) {
    if (!reader) throw new Error('RepoMirror: reader (DidPeer-shaped) required')
    if (!store) throw new Error('RepoMirror: store (OverlayRepoStore) required')
    this._reader = reader
    this._store = store
    this._onSync = onSync
    this._running = false
    this._again = false
    this._closed = false
  }

  // Wire the event-driven trigger and do the initial sync. Returns after the first pass.
  async start () {
    // DidPeer invokes this._onHead(head) on every NEW verified head (did-peer.js:160). Chain it so a
    // replicated firehose commit re-syncs the files — no polling timer (P2P: event-driven).
    const prev = this._reader._onHead
    this._reader._onHead = (head) => { try { if (prev) prev(head) } catch {} ; this.sync().catch(() => {}) }
    await this.sync()
    return this
  }

  // Snapshot-reconcile the local files to the reader's verified repo: write new/changed records,
  // delete records no longer present. Idempotent + byte-stable (skips a record whose local CID
  // already matches), and coalesced so concurrent head-advances collapse into one trailing pass.
  async sync () {
    if (this._closed) return null
    if (this._running) { this._again = true; return null }
    this._running = true
    try {
      if (!this._reader.head) return null // no verified head yet — nothing to converge to
      const want = new Map() // "coll/rkey" -> { collection, rkey, value, cid }
      for (const coll of await this._reader.collections()) {
        for (const rec of await this._reader.listRecords(coll)) {
          const rkey = rec.uri.slice(rec.uri.lastIndexOf('/') + 1)
          want.set(coll + '/' + rkey, { collection: coll, rkey, value: rec.value, cid: rec.cid })
        }
      }
      let wrote = 0
      for (const w of want.values()) {
        const cur = await this._store.getRaw(w.collection, w.rkey)
        if (cur && cidForBytesSync(cur.bytes).toString() === w.cid) continue // already byte-identical
        await this._store.put(w.collection, w.rkey, w.value)
        wrote++
      }
      let removed = 0
      // OverlayRepoStore.records() -> [{ key: "collection/rkey", bytes, record }] (read-view shape).
      for (const r of await this._store.records()) {
        if (!want.has(r.key)) {
          const i = r.key.indexOf('/')
          await this._store.del(r.key.slice(0, i), r.key.slice(i + 1))
          removed++
        }
      }
      const summary = { total: want.size, wrote, removed }
      if (this._onSync) try { this._onSync(summary) } catch {}
      return summary
    } finally {
      this._running = false
      if (this._again && !this._closed) { this._again = false; this.sync().catch(() => {}) }
    }
  }

  close () {
    this._closed = true
    try { this._reader._onHead = null } catch {}
  }
}

// dagCbor imported only to keep the byte-exactness intent explicit at call sites that pre-encode;
// RepoMirror itself passes decoded values to OverlayRepoStore.put (which dag-cbor encodes verbatim —
// canonical/deterministic, so the re-encoded file CID matches the signer's record CID).
export { dagCbor }