favela-web / lib / label-firehose.js

// label-firehose.js — com.atproto.label.subscribeLabels: the LIVE-PUSH surface bsky's AppView/Ozone
// ingests (lens export §3.6, moderation half; the push counterpart to lib/labeler-export.js's queryLabels).
//
// The atproto wire format (matches atproto-store/lib/firehose.js's subscribeRepos framing): each WS binary
// message = dag-cbor(header {t:'#labels', op:1}) ++ dag-cbor(body {seq, labels:[label…]}). Labels are the
// com.atproto.label.defs#label objects with `sig` as raw bytes (DAG-CBOR carries bytes natively — NOT the
// JSON { $bytes } form queryLabels uses). A subscriber connects with ?cursor=<lastSeq>; the server replays
// from there, then live-tails. seq is a 1-based monotonic index over the labeler's feed.
import * as dagCbor from '@ipld/dag-cbor'
import { decodeLabel } from 'atproto-identity/lib/labels.js'

const LABELS_HEADER = dagCbor.encode({ t: '#labels', op: 1 })
const ERROR_HEADER = dagCbor.encode({ op: -1 })

function concat (a, b) { const out = new Uint8Array(a.length + b.length); out.set(a, 0); out.set(b, a.length); return out }

// One subscribeLabels frame: header ++ dag-cbor({ seq, labels }). labels = signed label objects (sig bytes).
export function labelFrame (seq, labels) { return concat(LABELS_HEADER, dagCbor.encode({ seq, labels })) }

// An error frame ({ op:-1 } ++ { error, message }) — e.g. FutureCursor when the client's cursor is ahead.
export function errorFrame (error, message = '') { return concat(ERROR_HEADER, dagCbor.encode({ error, message })) }

// Decode a subscribeLabels frame (for tests / a consumer): → { t, body } or { op:-1, body } or null.
export function decodeLabelsFrame (bytes) {
  if (bytes.length >= LABELS_HEADER.length && Buffer.from(bytes.subarray(0, LABELS_HEADER.length)).equals(Buffer.from(LABELS_HEADER))) {
    return { t: '#labels', body: dagCbor.decode(bytes.subarray(LABELS_HEADER.length)) }
  }
  if (bytes.length >= ERROR_HEADER.length && Buffer.from(bytes.subarray(0, ERROR_HEADER.length)).equals(Buffer.from(ERROR_HEADER))) {
    return { op: -1, body: dagCbor.decode(bytes.subarray(ERROR_HEADER.length)) }
  }
  return null
}

export class LabelFirehose {
  // core: a hypercore of encoded labels (the labeler's feed — each block = dag-cbor(signed label)). Needs
  // { length, get(i), on/off('append'), ready }.
  constructor (core) { this.core = core }

  async ready () { if (this.core.ready) await this.core.ready(); return this }

  // Serve one subscriber. send(bytes) writes a binary WS message. cursor = the last seq the client holds
  // (0/absent → from the start). Replays [cursor, head) then live-tails on append. Runs are serialized +
  // coalesced (append during a replay re-runs after) so no frame is skipped/duplicated. If cursor is past
  // the head, sends a FutureCursor error frame first. Returns stop() to detach the append listener.
  serve (send, { cursor = 0 } = {}) {
    let closed = false
    let running = false
    let pending = false
    let i = Math.max(0, Number(cursor) || 0) // seq = i+1, so start index = cursor
    if (i > this.core.length) { try { send(errorFrame('FutureCursor', 'cursor is past the labeler head')) } catch {} ; i = this.core.length }
    const pump = async () => {
      if (running) { pending = true; return }
      running = true
      try {
        while (!closed && i < this.core.length) {
          const idx = i++
          try { send(labelFrame(idx + 1, [decodeLabel(await this.core.get(idx))])) } catch {}
        }
      } finally { running = false }
      if (!closed && pending) { pending = false; return pump() }
    }
    const onAppend = () => { pump().catch(() => {}) }
    this.core.on('append', onAppend)
    pump().catch(() => {})
    return () => { if (closed) return; closed = true; this.core.off('append', onAppend) }
  }
}

export default LabelFirehose