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

/**
 * OrgEventLog — thin wrapper around a single Hypercore that stores
 * append-only org lifecycle events (created / updated / deleted).
 *
 * Sister to RepoEventLog. The two have the exact same architecture:
 * single-writer Hypercore on the primary, read-only replicas on
 * mirrors. Each entry carries two per-org autobase keys so a mirror
 * replicates a full org (#37 per-org-autobase model):
 *   - `orgBootstrapKey`: the org's RECORD autobase → mirror mountOrg()'s it
 *   - `issueBootstrapKey`: the per-org ISSUE tracker autobase → mirror binds
 *     it to the primary's root instead of lazy-creating a sibling.
 *
 * Writer (primary): opens the core under `corestore.namespace('org-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, or from the deployment bundle). 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 ("bind-if-missing"). No cursor state is persisted.
 *
 * GC / compaction is deferred — at v3 scale (a handful of orgs) the
 * log grows at ~50 B per create event, negligible for years.
 */

import c from 'compact-encoding'

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

export class OrgEventLog {
  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('org-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('org-event-log: not writable')
    const buf = c.encode(entrySchema, {
      op: e.op | 0,
      name: e.name,
      ts: e.ts || Date.now(),
      issueBootstrapKey: e.issueBootstrapKey || null,
      orgBootstrapKey: e.orgBootstrapKey || 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('org-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 {}
  }
}