// FleetRelay — the egress membrane (§11 / the "edge is a little-heap relay"). PROTOTYPE.
//
// The atproto analogue of hyperident-gateway: HTTP on one side, Hypercore on the other. Each
// fleet PDS's firehose is a Hypercore (FirehoseSequencer); the relay REPLICATES those
// (Track 3), MERGES them with a FleetSequencer (the fan-in that keeps the atproto seq stable
// + monotonic even as sources are read in any order), and PROJECTS the aggregate as an HTTP
// com.atproto.sync.subscribeRepos firehose for bsky.network to crawl. So the whole fleet —
// including NAT'd members reachable only over Hyperswarm — federates through one endpoint.
//
// This class is the aggregation core: given source firehose Hypercores (obtained however —
// in-process, or replicated over the fleet swarm), it produces ONE well-behaved firehose.
// The HTTP subscribeRepos server + the swarm source-replication wrap around it.
import b4a from 'b4a'
import { FleetSequencer } from './fleet-sequencer.js'
import { restampFrame } from 'atproto-store/lib/firehose.js'
export class FleetRelay {
// `seqBase` offsets every EMITTED seq (emitted = seqBase + internal). This is load-bearing for
// relay recovery: atproto relays (indigo) keep a strictly-forward cursor per host and NEVER reset
// it — if a redeploy resets our internal seq below the relay's stored cursor, the relay treats us
// as a "future cursor", parks Idle, and never reads again. The ONLY recovery is PDS-side: push our
// emitted seq ABOVE the relay's stored cursor (read via com.atproto.sync.getHostStatus). seqBase is
// persisted (via onSeqBase) so the emitted sequence stays durable + monotonic across redeploys —
// the atproto firehose contract. `setSeqBase(n)` only ever raises it.
constructor (aggregateCore, { seqBase = 0, onSeqBase = null } = {}) {
this.seq = new FleetSequencer(aggregateCore) // the merged firehose (its own Hypercore); INTERNAL seq
this._sources = new Map() // sourceId -> { core, onappend }
this._seqBase = Number(seqBase) || 0
this._onSeqBase = onSeqBase // async|sync (n) => persist
}
async ready () { await this.seq.ready(); return this }
get seqBase () { return this._seqBase }
get length () { return this._seqBase + this.seq.length } // the latest EMITTED seq
// Raise the emitted-seq floor above `n` (monotonic — never lowers). Persists via onSeqBase.
async setSeqBase (n) {
n = Number(n) || 0
if (n <= this._seqBase) return this._seqBase
this._seqBase = n
if (this._onSeqBase) { try { await this._onSeqBase(n) } catch {} }
return this._seqBase
}
// Add a source PDS's firehose Hypercore. Drains its existing frames into the aggregate and
// live-follows new appends. Frames are deduped by (sourceId, index), so re-reads and
// reconnections never double-count — the aggregated seq stays stable (FleetSequencer).
async addSource (sourceId, firehoseCore) {
await firehoseCore.ready()
// A freshly-opened REPLICATED core reads length=0 until an explicit update() learns the peer's
// announced length — and 'append' only fires for blocks appended AFTER we connect, never for the
// frames that already existed. Without this update the initial _drain sees 0 frames and the peer's
// pre-existing #commit history never enters the aggregate (only a #sync re-baseline does), so the
// AppView never indexes it. update({wait:true}) syncs the length so _drain ingests the backlog.
try { await firehoseCore.update({ wait: true }) } catch { /* offline/first-connect — 'append' catches up */ }
const prev = this._sources.get(sourceId) // idempotent re-add: detach any prior listener so we never double-drain
if (prev) { try { prev.core.off('append', prev.onappend) } catch {} }
const onappend = () => this._drain(sourceId, firehoseCore).catch(() => {})
this._sources.set(sourceId, { core: firehoseCore, onappend })
await this._drain(sourceId, firehoseCore)
firehoseCore.on('append', onappend)
return this
}
// Remove a source whose firehose core was rebuilt/retired (new core key ⇒ new sourceId). Detaches the
// live-follow listener so a dead core stops being drained; already-aggregated frames stay in the
// monotonic seq (never rewritten). Returns true iff a source was removed.
removeSource (sourceId) {
const s = this._sources.get(sourceId)
if (!s) return false
try { s.core.off('append', s.onappend) } catch {}
this._sources.delete(sourceId)
return true
}
// Inject a synthetic frame (e.g. a #sync re-baseline event, Sync 1.1) into the aggregate.
// `key` dedups it (same key → ingested once), so re-emitting the same rev is idempotent. The
// frame's internal seq is re-stamped to the relay's global seq on subscribe(), like any frame.
async emit (key, frameBytes) {
if (this.seq.seqOf(key) != null) return false // already emitted
await this.seq.ingest({ key, data: b4a.toString(b4a.from(frameBytes), 'base64') })
return true
}
async _drain (sourceId, core) {
for (let i = 0; i < core.length; i++) {
const key = `${sourceId}:${i}`
if (this.seq.seqOf(key) != null) continue // already aggregated
const bytes = await core.get(i)
// Store the original frame (base64 in the json emit-log). On HTTP serve, the frame's
// internal `seq` is re-stamped to the relay's global seq — a projection detail handled
// by the subscribeRepos wrapper, not the aggregation.
await this.seq.ingest({ key, data: b4a.toString(b4a.from(bytes), 'base64') })
}
}
// The aggregated firehose: replay frames with EMITTED seq > cursor, then live-tail. The incoming
// `cursor` is in emitted space (seqBase + internal); translate to internal, and re-stamp each
// frame with its emitted seq so a downstream relay/AppView sees one monotonic, seqBase-lifted cursor.
async * subscribe (cursor = 0, opts) {
const internalCursor = cursor > this._seqBase ? cursor - this._seqBase : 0
for await (const { seq, key, data } of this.seq.subscribe(internalCursor, opts)) {
const emitted = this._seqBase + seq
yield { seq: emitted, key, bytes: restampFrame(new Uint8Array(b4a.from(data, 'base64')), emitted) }
}
}
async close () {
for (const { core, onappend } of this._sources.values()) core.off('append', onappend)
this._sources.clear()
await this.seq.close()
}
}