atproto-store / lib / sequencer.js

// FirehoseSequencer — the atproto firehose as a Hypercore.
//
// Each subscribeRepos message (a framed dag-cbor #commit/#identity/#account) is one
// append to a single-writer Hypercore; the block index + 1 IS the atproto `seq`
// (1-based, monotonic). `subscribe(cursor)` replays persisted frames with seq > cursor
// then live-tails — so a freshly-attached relay gets full history from cursor=0 (the
// fix for the non-archival-relay root cause, POSTMORTEM.md §Federation).
//
// This is both the atproto sequenced-firehose model and the P2P-native thesis: the
// firehose is not a side-channel, it's a replicable append-only log.

import { commitFrame, identityFrame, accountFrame } from './firehose.js'

export class FirehoseSequencer {
  constructor (core) { this.core = core } // a Hypercore, binary values

  async ready () { await this.core.ready(); return this }
  get length () { return this.core.length }        // == the latest seq
  async close () { await this.core.close() }

  // Append a pre-framed message; returns its 1-based seq.
  async append (frameBytes) {
    const seq = this.core.length + 1
    await this.core.append(frameBytes)
    return seq
  }

  // Sequence a #commit: assign the seq FIRST (frame body carries it), then append.
  async sequenceCommit (write, { time }) {
    const seq = this.core.length + 1
    const frame = await commitFrame(write, { seq, time })
    await this.core.append(frame)
    return { seq, frame }
  }

  async sequenceIdentity ({ did, handle, time }) {
    const seq = this.core.length + 1
    const frame = identityFrame({ seq, did, handle, time })
    await this.core.append(frame)
    return { seq, frame }
  }

  async sequenceAccount ({ did, active = true, status, time }) {
    const seq = this.core.length + 1
    const frame = accountFrame({ seq, did, active, status, time })
    await this.core.append(frame)
    return { seq, frame }
  }

  // Async iterator over { seq, bytes }: backfill frames with seq > cursor, then
  // live-tail new appends. cursor is the last seq the consumer holds (0/absent =
  // from the start). Pass an AbortSignal to stop (WS disconnect).
  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 bytes = await this.core.get(next)
        yield { seq: next + 1, bytes }
        next++
        continue
      }
      // caught up — wait for the next append (or abort)
      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
    }
  }
}