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

/**
 * ReleaseEventLog — append-only Hypercore of release-index mutations
 * (published / promoted / unpinned / deleted).
 *
 * Sister to RepoEventLog and OrgEventLog; identical architecture —
 * single-writer core on the origin, read-only replicas on mirrors.
 *
 * Writer (origin): opens the core under `corestore.namespace('release-events')`
 * as `{ name: 'log' }`. ReleaseManager appends one entry per mutation, at
 * exactly the point it already rewrites `_index.json`.
 *
 * Reader (mirror): opens the same core by key (from the origin's `dump-keys`
 * output, or from the deployment bundle). Opened by key means the local
 * instance is a read-only replica. Replication rides the shared Hyperswarm
 * topic via `corestore.replicate(conn)`, which hyper-svc's SwarmTransport
 * already calls on every connection — so this needs no transport of its own.
 *
 * WHY THIS EXISTS
 *
 * Release replication was previously driven by a 15-minute polling timer,
 * because the push path it would naturally have used (hyper-deploy's
 * `sendFileList` broadcast) was removed upstream in 0.4.1 — it was the thing
 * blowing secret-stream's 16 MiB atomic-write limit on a 100k-file dependency
 * cache. Only the solicited half (mirror asks for `_index.json`) was ever
 * built, so a poll stood in for an event that the origin already has: every
 * publish/promote/unpin/delete calls ReleaseManager._rescan(), which rewrites
 * the index. This log turns that existing moment into something a peer can
 * observe, which is what lets the timer be deleted rather than merely tuned.
 *
 * Entries are idempotent and carry no bytes — an entry is a WAKE-UP naming a
 * repo, not a payload. Reconciliation always re-reads the origin's authoritative
 * `_index.json` and diffs it, so a replayed, duplicated, or out-of-order entry
 * costs one extra scoped diff and can never produce a wrong result. Replaying
 * from seq 0 is therefore safe (the cursor is an optimisation, not correctness).
 *
 * GC / compaction is deferred, as with the sister logs: entries are ~60 B and
 * the fleet publishes on the order of tens of releases a day.
 */

import c from 'compact-encoding'

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

/** ReleaseEventLog op codes. Use these instead of bare 0/1/2/3. */
export const OP_PUBLISHED = 0
export const OP_PROMOTED = 1
export const OP_UNPINNED = 2
export const OP_DELETED = 3

export class ReleaseEventLog {
  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('release-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 one entry. Only valid for the writer (the origin).
   *
   * @param {object} e  { op, repo, version?, channel?, ts? }
   */
  async append (e) {
    if (!this.writable) throw new Error('release-event-log: not writable')
    const buf = c.encode(entrySchema, {
      op: e.op | 0,
      repo: e.repo || '',
      version: e.version || '',
      channel: e.channel || '',
      ts: e.ts || Date.now()
    })
    await this.core.append(buf)
  }

  /**
   * Async-iterate decoded entries. `opts` passes through to
   * hypercore.createReadStream — `{ live: true, start: N }` is the shape a
   * mirror uses to stream forever from its cursor.
   *
   * Malformed entries are skipped so one bad append cannot stall the stream.
   */
  async * entries (opts = {}) {
    if (!this.core) throw new Error('release-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 () {
    if (this._activeStreams) {
      for (const s of this._activeStreams) {
        try { s.destroy() } catch {}
      }
      this._activeStreams.clear()
    }
    try { await this.core?.close() } catch {}
  }
}