// FleetSequencer — the EDGE firehose sequencer for a fleet (§11, "the one genuinely hard
// sub-problem"). PROTOTYPE.
//
// The problem: the fleet's internal aggregation (a shared multi-writer autobase) can
// RELINEARIZE — autobase reorders history as the causal graph changes. But atproto's
// firehose `seq` MUST be append-only monotonic: relays track a cursor and reconnect with
// "give me everything after seq N". A naïve autobase-as-firehose (seq = view index) breaks
// that contract the moment linearization shifts an earlier event.
//
// The resolution (doc §11): the EDGE is the sequencer. It assigns the bsky-facing seq at
// EMIT time and never rewrites it. This class owns a single-writer, append-only emit log
// (its OWN Hypercore — monotonic by construction, exactly like the per-instance
// FirehoseSequencer). It ingests events from the reorderable aggregation and dedupes by a
// stable event key, appending each NEW event exactly once in first-seen order. Because the
// emit log is append-only and an already-seen key never re-appends, an event's seq is
// STABLE no matter how the autobase reorders underneath. bsky never sees the autobase; it
// sees this well-behaved firehose. The emit log itself IS the durable cursor map (the
// seen-map is rebuilt from it on open), so seq survives restarts.
//
// `key` is the stable identity of a logical firehose event — e.g. a #commit's commit CID,
// or `${did}#${rev}`; a synthetic stable id for #identity/#account. Two writers must never
// mint the same key for different events (content-addressing gives this for free).
export class FleetSequencer {
constructor (core) { this.core = core } // Hypercore, valueEncoding 'json': { key, data }
// Rebuild the seen-map (key -> seq) from the durable emit log — so an already-assigned
// seq survives a restart and dedup keeps working across process lifetimes.
async ready () {
await this.core.ready()
this.seen = new Map()
for (let i = 0; i < this.core.length; i++) {
const block = await this.core.get(i)
this.seen.set(block.key, i + 1)
}
return this
}
get length () { return this.core.length } // == the latest emitted seq
async close () { await this.core.close() }
seqOf (key) { return this.seen.has(key) ? this.seen.get(key) : null }
// Ingest one aggregation event. If its key was already emitted, it KEEPS its original
// seq and does NOT append — this is precisely what makes seq stable under relinearization
// and idempotent under re-delivery. A genuinely new key gets the next seq.
async ingest ({ key, data }) {
if (key == null) throw new Error('FleetSequencer.ingest: event key required')
const existing = this.seen.get(key)
if (existing != null) return { seq: existing, deduped: true }
const seq = this.core.length + 1
await this.core.append({ key, data })
this.seen.set(key, seq)
return { seq, deduped: false }
}
async ingestAll (events) {
const out = []
for (const e of events) out.push(await this.ingest(e))
return out
}
// subscribe(cursor): replay emitted events with seq > cursor, then live-tail. The stable,
// append-only firehose a relay consumes — identical replay regardless of ingest order.
async * subscribe (cursor = 0, { signal } = {}) {
let next = Math.max(0, cursor | 0) // seq > cursor <=> index >= cursor
while (true) {
if (signal && signal.aborted) return
if (next < this.core.length) {
const block = await this.core.get(next)
yield { seq: next + 1, key: block.key, data: block.data }
next++
continue
}
await new Promise((resolve) => {
const onAppend = () => { cleanup(); resolve() }
const onAbort = () => { cleanup(); resolve() }
const cleanup = () => {
this.core.off('append', onAppend)
if (signal) signal.removeEventListener('abort', onAbort)
}
this.core.on('append', onAppend)
if (signal) signal.addEventListener('abort', onAbort, { once: true })
})
if (signal && signal.aborted) return
}
}
}