favela-web / test / label-firehose.test.mjs

// favela-web subscribeLabels firehose — the atproto wire framing + replay/live-tail, over a mock label
// feed with REAL signed labels. Run: node test/label-firehose.test.mjs
import { LabelFirehose, decodeLabelsFrame, labelFrame } from '../lib/label-firehose.js'
import { buildLabel, signLabel, verifyLabel, encodeLabel } from 'atproto-identity/lib/labels.js'
import { keyFromPrivate } from 'atproto-identity/lib/atproto-identity.js'
import b4a from 'b4a'

const assert = (c, m) => { if (!c) { console.error('FAIL:', m); process.exit(1) } console.log('  ok -', m) }
const sleep = (ms) => new Promise((r) => setTimeout(r, ms))

const key = keyFromPrivate(b4a.from('22'.repeat(32), 'hex'))
const LABELER = 'did:web:favela.example.net'
const mkLabel = (uri, val, cts) => encodeLabel(signLabel(buildLabel({ src: LABELER, uri, val, cts }), key.privateKey))

// mock hypercore of encoded labels
function mockCore () {
  const blocks = []; const listeners = new Set()
  return {
    get length () { return blocks.length },
    async get (i) { return blocks[i] },
    on (ev, cb) { if (ev === 'append') listeners.add(cb) },
    off (ev, cb) { if (ev === 'append') listeners.delete(cb) },
    ready () {},
    _push (bytes) { blocks.push(bytes); for (const cb of [...listeners]) cb() }
  }
}

// ── frame codec round-trips (labels' sig stays raw bytes over DAG-CBOR) ──
{
  const l = signLabel(buildLabel({ src: LABELER, uri: 'at://did:plc:a/app.bsky.feed.post/1', val: 'spam', cts: '2026-07-15T00:00:00.000Z' }), key.privateKey)
  const dec = decodeLabelsFrame(labelFrame(7, [l]))
  assert(dec.t === '#labels' && dec.body.seq === 7, 'labelFrame → {t:#labels, body.seq}')
  assert(dec.body.labels.length === 1 && dec.body.labels[0].val === 'spam', 'frame carries the label')
  assert(verifyLabel(dec.body.labels[0], key.didKey), 'the framed label still verifies (sig survived DAG-CBOR as raw bytes)')
}

// ── replay from cursor 0 → all frames, seqs 1..N ──
const core = mockCore()
core._push(mkLabel('at://did:plc:a/app.bsky.feed.post/1', 'spam', '2026-07-15T00:00:00.000Z'))
core._push(mkLabel('at://did:plc:b/app.bsky.feed.post/2', '!hide', '2026-07-15T00:01:00.000Z'))
const fh = await new LabelFirehose(core).ready()

const got0 = []
const stop0 = fh.serve((bytes) => got0.push(decodeLabelsFrame(bytes)), { cursor: 0 })
await sleep(50)
assert(got0.length === 2 && got0[0].body.seq === 1 && got0[1].body.seq === 2, 'replay from cursor 0 → seqs 1,2')
assert(got0[0].body.labels[0].val === 'spam' && got0[1].body.labels[0].val === '!hide', 'replayed labels in order')

// ── live-tail: a label appended AFTER subscribe is pushed ──
core._push(mkLabel('at://did:plc:c/app.bsky.feed.post/3', 'nsfw', '2026-07-15T00:02:00.000Z'))
await sleep(50)
assert(got0.length === 3 && got0[2].body.seq === 3 && got0[2].body.labels[0].val === 'nsfw', 'LIVE-TAIL: an appended label is pushed as seq 3')
stop0()

// ── replay from a cursor mid-stream → only newer seqs ──
const got1 = []
const stop1 = fh.serve((bytes) => got1.push(decodeLabelsFrame(bytes)), { cursor: 1 })
await sleep(50)
assert(got1.length === 2 && got1[0].body.seq === 2 && got1[1].body.seq === 3, 'cursor=1 replays from seq 2 (skips already-seen)')
stop1()

// ── a cursor past the head → FutureCursor error frame ──
const gotErr = []
const stopE = fh.serve((bytes) => gotErr.push(decodeLabelsFrame(bytes)), { cursor: 99 })
await sleep(50)
assert(gotErr.some((f) => f && f.op === -1 && f.body.error === 'FutureCursor'), 'cursor past head → FutureCursor error frame')
stopE()

// ── stop() detaches: no frames after stop ──
const got2 = []
const stop2 = fh.serve((bytes) => got2.push(bytes), { cursor: 3 })
await sleep(30); stop2()
const n = got2.length
core._push(mkLabel('at://did:plc:d/app.bsky.feed.post/4', 'spam', '2026-07-15T00:03:00.000Z'))
await sleep(50)
assert(got2.length === n, 'stop() detaches the live-tail (no frames after stop)')

console.log('\nLABEL-FIREHOSE OK: subscribeLabels framing + replay-from-cursor + live-tail + FutureCursor + stop')