pear-git-v3 / lib / services / repo-event-log.js

/**
 * RepoEventLog — thin wrapper around a single Hypercore that stores
 * append-only repo lifecycle events (created / updated / deleted).
 *
 * Writer (primary): opens the core under `corestore.namespace('repo-events')`
 * as `{ name: 'log' }`. Appends one encoded `entry` per lifecycle event.
 *
 * Reader (mirror): opens the same core by key (from the primary's
 * `dump-keys` output). Because it's opened by key, the local instance is
 * a read-only replica — mirrors cannot append. Replication happens via
 * the shared Hyperswarm topic and `corestore.replicate(conn)`, which
 * hyper-svc's SwarmTransport already calls on every connection.
 *
 * Entries are idempotent: a mirror can replay the whole log from seq 0
 * without harm ("clone-if-missing, fetch-if-present"). No cursor state
 * is persisted.
 *
 * GC / compaction is deferred (see plan O-11). Hypercore has no native
 * truncation; at v3 scale the log grows at ~100 B × events/day, which
 * is negligible for years.
 */

import c from 'compact-encoding'

import { entry as entrySchema } from './repo-event-log-messages.js'

export class RepoEventLog {
  constructor ({ corestore, key = null }) {
    this._corestore = corestore
    this._givenKey = key
      ? (typeof key === 'string' ? Buffer.from(key, 'hex') : key)
      : null
    this.core = null
  }

  async ready () {
    if (this.core) return
    this.core = this._givenKey
      ? this._corestore.get({ key: this._givenKey })
      : this._corestore.namespace('repo-events').get({ name: 'log' })
    await this.core.ready()
  }

  get keyHex () {
    return this.core && this.core.key ? this.core.key.toString('hex') : null
  }

  get writable () {
    return !!(this.core && this.core.writable)
  }

  get length () {
    return this.core ? this.core.length : 0
  }

  /**
   * Append a new entry to the log. Only valid for the writer.
   */
  async append (e) {
    if (!this.writable) throw new Error('repo-event-log: not writable')
    const buf = c.encode(entrySchema, {
      op: e.op | 0,
      name: e.name,
      head: e.head || '',
      ts: e.ts || Date.now(),
      issueBootstrapKey: e.issueBootstrapKey || null
    })
    await this.core.append(buf)
  }

  /**
   * Async-iterate decoded entries from the log. `opts` passes through to
   * hypercore.createReadStream — `{ live: true, start: 0 }` is the usual
   * shape for a mirror that wants to stream forever from the beginning.
   *
   * Malformed entries are silently skipped so a single bad append can't
   * stall the stream.
   */
  async * entries (opts = {}) {
    if (!this.core) throw new Error('repo-event-log: call ready() first')
    const stream = this.core.createReadStream(opts)
    this._activeStreams = this._activeStreams || new Set()
    this._activeStreams.add(stream)
    try {
      for await (const buf of stream) {
        try {
          yield c.decode(entrySchema, buf)
        } catch {
          // skip undecodable entries (forward-compat / corruption safety)
        }
      }
    } finally {
      this._activeStreams.delete(stream)
    }
  }

  async close () {
    // Destroy any in-flight read streams so async iterators can unwind.
    if (this._activeStreams) {
      for (const s of this._activeStreams) {
        try { s.destroy() } catch {}
      }
      this._activeStreams.clear()
    }
    try { await this.core?.close() } catch {}
  }
}