/**
* RepoFollower — read-only single-repo follower.
*
* Counterpart to RepoPuller (services/repo-mirror.js). RepoPuller is
* bulk-mode (one origin, all repos, writer-enrollment via mirror flow).
* RepoFollower is single-resource (one origin, one repo, NEVER enrolls
* as a writer).
*
* Lifecycle
* ---------
* start():
* 1. Open the source's RepoEventLog by key (corestore replicates
* on first matching swarm connection).
* 2. Open the per-repo HyperIssues tracker via FollowRegistry
* (namespace `follow:<srvKey>:repo:<name>`).
* 3. addRemote('follow:<srvKey>', sourceServerKey) for git fetches.
* 'primary' is reserved for the mirror flow — followers use a
* distinct remote name so they coexist with a mirror's origin.
* 4. Join the source's swarm topic with client:true. Corestore
* replication runs on every new connection via the existing
* SwarmTransport wiring.
* 5. Initial git pull. Idempotent: clone-if-missing,
* fetch-if-present.
* 6. If repoEventLogKey was provided, start the live-read loop and
* react to any entry where entry.name === this followed name.
*
* close():
* Stop the live loop, leave the swarm topic, close the event log.
* Per the never-delete-storage rule, on-disk cores + git-repo bytes
* are preserved. Re-follow restores from disk.
*
* Read-only invariant
* -------------------
* RepoFollower never calls `addWriter` on any autobase. The follower's
* tracker.writable will remain false. If a follower needs to become a
* writer, that's a different flow (mirror enrollment via
* `pgh device join`) — by design.
*
* Cursor file (optional, opts.cursorPath):
* { '<sourceServerKey>|<repoName>': <lastProcessedSeq> }
*
* Sharing the file across many RepoFollower instances is safe — keys
* are per-(source, name).
*/
import b4a from 'b4a'
import fs from '#node:fs'
import path from '#node:path'
import crypto from 'hypercore-crypto'
import { RepoEventLog } from './repo-event-log.js'
export class RepoFollower {
constructor (opts = {}) {
if (!opts.corestore) throw new Error('RepoFollower requires { corestore }')
if (!opts.swarm) throw new Error('RepoFollower requires { swarm }')
if (!opts.sync) throw new Error('RepoFollower requires { sync }')
if (!opts.followRegistry) throw new Error('RepoFollower requires { followRegistry }')
if (!opts.sourceServerKey) throw new Error('RepoFollower requires { sourceServerKey }')
if (!opts.repoName) throw new Error('RepoFollower requires { repoName }')
this._corestore = opts.corestore
this._swarm = opts.swarm
this._sync = opts.sync
this._followRegistry = opts.followRegistry
this._sourceServerKey = normalizeHex(opts.sourceServerKey)
this._repoName = opts.repoName
this._issueBootstrapKey = opts.issueBootstrapKey
? normalizeHex(opts.issueBootstrapKey)
: null
this._repoEventLogKey = opts.repoEventLogKey
? normalizeHex(opts.repoEventLogKey)
: null
this._topicSalt = opts.topicSalt || 'pear-git-v3'
this._cursorPath = opts.cursorPath || null
this._log = opts.log || { info: () => {}, warn: () => {}, error: () => {} }
this._onProgress = opts.onProgress || null
this._remoteName = 'follow:' + this._sourceServerKey
this._cursorKey = this._sourceServerKey + '|' + this._repoName
this._started = false
this._stopped = false
this._tracker = null // HyperIssues (read-only)
this._eventLog = null // RepoEventLog (read-only)
this._discovery = null // swarm topic membership handle
this._inflightPull = false
this._pendingPull = false // a live entry arrived while pull in flight
this._cursor = this._loadCursor()
}
_loadCursor () {
if (!this._cursorPath) return {}
try {
const raw = fs.readFileSync(this._cursorPath, 'utf8')
const parsed = JSON.parse(raw)
if (parsed && typeof parsed === 'object') return parsed
} catch {}
return {}
}
_saveCursor () {
if (!this._cursorPath) return
try {
fs.mkdirSync(path.dirname(this._cursorPath), { recursive: true })
const tmp = this._cursorPath + '.tmp'
fs.writeFileSync(tmp, JSON.stringify(this._cursor, null, 2))
fs.renameSync(tmp, this._cursorPath)
} catch {}
}
async start () {
if (this._started) return
this._started = true
// 1. Open read-only tracker via FollowRegistry (per-source namespace).
if (this._issueBootstrapKey) {
try {
this._tracker = await this._followRegistry.openRepo(
this._sourceServerKey, this._repoName, this._issueBootstrapKey
)
} catch (err) {
this._log.warn('[follow] tracker open failed:', err && err.message)
}
}
// 2. Register sync remote so sync.pull(remoteName, repoName) routes.
if (!this._sync.remotes.has(this._remoteName)) {
this._sync.addRemote(this._remoteName, this._sourceServerKey)
}
// 3. Join the source's swarm topic (client-only — followers don't
// serve, they read). Corestore.replicate fires on each new
// connection automatically.
try {
const topic = computeTopic(this._sourceServerKey, this._topicSalt)
this._discovery = this._swarm.join(topic, { server: false, client: true })
try { await this._discovery.flushed() } catch {}
} catch (err) {
this._log.warn('[follow] swarm.join failed:', err && err.message)
}
// 4. Initial git pull (cold or warm). Idempotent.
this._tryPull().catch((err) => {
this._log.warn('[follow] initial pull failed:', err && err.message)
})
// 5. If we have an event log key, start the live loop for ongoing
// push notifications. Without it, follower still works but only
// catches up on initial start + manual re-follow.
if (this._repoEventLogKey) {
try {
this._eventLog = new RepoEventLog({
corestore: this._corestore,
key: this._repoEventLogKey
})
await this._eventLog.ready()
} catch (err) {
this._log.warn('[follow] event log open failed:', err && err.message)
this._eventLog = null
}
if (this._eventLog) {
this._runLoop().catch((err) => {
this._log.error('[follow] event loop exited:', err && err.message)
})
}
}
}
async _runLoop () {
try { await this._eventLog.core.update({ wait: true }) } catch {}
const lastSeq = this._cursor[this._cursorKey]
const startSeq = typeof lastSeq === 'number' && lastSeq >= 0 ? lastSeq + 1 : 0
let seq = startSeq
for await (const entry of this._eventLog.entries({ live: true, start: startSeq })) {
if (this._stopped) break
// Filter to OUR repo only. RepoEventLog is one log for ALL repos
// on the source — we ignore everything else (privacy note in
// spec/modules/follow-payload.md).
if (entry && entry.name === this._repoName && entry.op !== 2 /* deleted */) {
if (this._inflightPull) {
// Origin pushed during our pull — mark for re-pull when done.
this._pendingPull = true
} else {
this._tryPull(seq).catch((err) => {
this._log.warn('[follow] live pull failed:', err && err.message)
})
}
}
// Cursor always advances — we've SEEN the entry; whether it was
// for our repo or not, replaying it on the next boot is harmless
// (no-op if not ours; idempotent fetch if ours).
const prev = this._cursor[this._cursorKey]
if (typeof prev !== 'number' || seq > prev) {
this._cursor[this._cursorKey] = seq
this._saveCursor()
}
seq++
}
}
async _tryPull (seq) {
if (this._stopped) return
this._inflightPull = true
let ok = false, err = null
try {
await this._sync.pull(this._remoteName, this._repoName)
ok = true
} catch (e) {
err = e
} finally {
this._inflightPull = false
try { if (this._onProgress) this._onProgress({ name: this._repoName, seq, ok, err }) } catch {}
// Re-pull if the loop saw newer entries during this pull.
if (this._pendingPull && !this._stopped) {
this._pendingPull = false
this._tryPull().catch(() => {})
}
}
}
status () {
return {
sourceServerKey: this._sourceServerKey,
repoName: this._repoName,
hasTracker: !!this._tracker,
hasEventLog: !!this._eventLog,
inflightPull: this._inflightPull,
cursor: this._cursor[this._cursorKey] !== undefined ? this._cursor[this._cursorKey] : null
}
}
async close () {
if (this._stopped) return
this._stopped = true
try { if (this._discovery) await this._discovery.destroy() } catch {}
try { if (this._eventLog) await this._eventLog.close() } catch {}
// FollowRegistry owns the tracker lifecycle — we don't close it
// here. close() on FollowRegistry on daemon shutdown closes all.
this._discovery = null
this._eventLog = null
}
}
// ── helpers ──
function normalizeHex (v) {
if (typeof v === 'string') return v.toLowerCase()
if (v && typeof v.toString === 'function') return v.toString('hex')
throw new Error('expected hex string or buffer')
}
function computeTopic (sourceServerKeyHex, salt) {
// Match pear-git-v3's own swarm topic derivation (lib/pear-git.js).
return crypto.hash(b4a.concat([
b4a.from(sourceServerKeyHex, 'hex'),
b4a.from(salt)
]))
}