pear-git-v3 / lib / services / org-mirror.js

/**
 * OrgPuller — sister to RepoPuller, but for orgs.
 *
 * Tracks a remote origin peer's OrgEventLog and, for each org, mounts
 * the two per-org autobases the entry advertises (#37 per-org-autobase
 * model — there is no single shared orgs autobase anymore):
 *
 *   - orgBootstrapKey  → `orgs.mountOrg(name, key)` so the org RECORD
 *     autobase replicates over the shared swarm (the mirror sees the org).
 *     Org issues live IN this autobase (linked-core #42), so mounting the
 *     record autobase replicates the org's issues too — there is no
 *     separate per-org issue tracker to bind.
 *
 * No git pull — orgs are pure metadata.
 *
 * Boots only when the orchestrator is configured with
 * `opts.origin = { serverKey, orgEventLogKey }`.
 *
 * What it does
 * ------------
 * - Opens the origin's org-event-log by key — corestore replicates it
 *   automatically over the shared Hyperswarm topic.
 * - Streams decoded entries from seq 0 forever (live).
 * - Mounts/binds per the keys above.
 *
 * Idempotent on replay — mountOrg is a no-op for an already-mounted name.
 */

import b4a from 'b4a'

import { parseSchemaVersionEntry } from './schema-guard.js'

export class OrgPuller {
  constructor ({ issueRegistry, orgs, log, supportedVersion, onBehind, onMount }) {
    if (!issueRegistry) throw new Error('OrgPuller requires { issueRegistry }')
    if (!log) throw new Error('OrgPuller requires { log }')

    this._issueRegistry = issueRegistry
    this._orgs = orgs || null // HyperOrgs — mount org RECORD autobases
    // Fired with the org name after a NEW org RECORD autobase is mounted, so
    // the daemon can re-announce mirror-writer keys (post-boot enrollment).
    this._onMount = typeof onMount === 'function' ? onMount : null
    this._log = log
    this._started = false
    this._stopped = false

    // In-band schema guard (#52 4.4): the highest schema version this code
    // supports. If the origin's log declares a NEWER version, we're stale —
    // interrupt (Keet's unsupportedABI): stop binding stale orgs + signal.
    this._supportedVersion = typeof supportedVersion === 'number' ? supportedVersion : Infinity
    this._onBehind = typeof onBehind === 'function' ? onBehind : null
    this._behind = false

    // Hook for tests — fires { name, ok, err? } per processed entry.
    this._onProgress = null
  }

  async start () {
    if (this._started) return
    this._started = true
    this._runLoop().catch((err) => {
      // eslint-disable-next-line no-console
      console.error('[org-mirror] read loop exited:', err && err.message)
    })
  }

  async _runLoop () {
    // Prime the core: wait for it to know its real length from peers.
    // Without this, the live read stream might start with length 0 and
    // hang because the core never emits an 'append' event for blocks
    // that arrived before subscription. Same fix as RepoPuller.
    try {
      await this._log.core.update({ wait: true })
    } catch {}

    for await (const entry of this._log.entries({ live: true, start: 0 })) {
      if (this._stopped) break
      await this._process(entry)
    }
  }

  async _process (entry) {
    if (!entry || !entry.name) return

    // Schema-version marker (Keet's interrupt): if the origin declares a
    // schema NEWER than we support, this node is running stale code. Flag it
    // loudly and stop binding orgs — don't replicate/serve a newer format we
    // can't represent. The marker carries no org to mount, so always return.
    const declaredVersion = parseSchemaVersionEntry(entry)
    if (declaredVersion !== null) {
      if (declaredVersion > this._supportedVersion && !this._behind) {
        this._behind = true
        // eslint-disable-next-line no-console
        console.error('[org-mirror][schema-guard] origin declares schema v' + declaredVersion +
          ' but this node supports v' + this._supportedVersion +
          ' — INTERRUPT: refusing to bind orgs until this node is upgraded')
        if (this._onBehind) { try { this._onBehind({ declared: declaredVersion, supported: this._supportedVersion }) } catch {} }
      }
      return
    }

    if (this._behind) return // interrupted — don't bind stale orgs after a newer-schema marker
    if (entry.op === 2) return // 'deleted' — never-delete-storage rule
    let ok = false
    let err = null
    try {
      // Mount the org RECORD autobase so the org itself replicates.
      if (entry.orgBootstrapKey && this._orgs && typeof this._orgs.mountOrg === 'function') {
        const orgKeyHex = typeof entry.orgBootstrapKey === 'string'
          ? entry.orgBootstrapKey
          : b4a.toString(entry.orgBootstrapKey, 'hex')
        await this._orgs.mountOrg(entry.name, orgKeyHex)
      }
      ok = true
    } catch (e) {
      err = e
      // eslint-disable-next-line no-console
      console.error('[org-mirror] bind/open failed for', entry.name, '-', e && e.message)
    } finally {
      try { if (this._onProgress) this._onProgress({ name: entry.name, ok, err }) } catch {}
      if (ok) { try { if (this._onMount) this._onMount(entry.name) } catch {} }
    }
  }

  async close () {
    this._stopped = true
    try { await this._log.close() } catch {}
  }
}