favela-web / test / subscribe-labels-ws.test.mjs

// favela-web subscribeLabels over a REAL WebSocket (bare-ws) — the transport the framing test mocks.
// A real Labeler feed → LabelFirehose → bare-ws server (shared HTTP server via 'upgrade') → a bare-ws
// client that decodes + VERIFIES the labels off the wire, incl. live-tail. Run: bare test/subscribe-labels-ws.test.mjs
import '../lib/bare-globals.js'
import fs from 'bare-fs'
import http from 'bare-http1'
import bareWs from 'bare-ws'
import Corestore from 'corestore'
import crypto from 'hypercore-crypto'
import b4a from 'b4a'
import { Labeler } from '../lib/labeler.js'
import { LabelFirehose, decodeLabelsFrame } from '../lib/label-firehose.js'
import { verifyLabel } from 'atproto-identity/lib/labels.js'

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 PORT = 4058
const dir = '/tmp/favela-ws-' + Date.now(); fs.mkdirSync(dir, { recursive: true })

// real labeler feed with two signed labels
const store = new Corestore(dir)
const labeler = await new Labeler(store, { privateKey: crypto.hash(b4a.from('favela-ws-test-seed')), src: 'did:web:favela.test', name: 'lf' }).ready()
await labeler.label('at://did:plc:a/app.bsky.feed.post/1', 'spam')
await labeler.label('at://did:plc:b/app.bsky.feed.post/2', '!hide')
const firehose = await new LabelFirehose(labeler.core).ready()

// server: bare-http1 + bare-ws over the 'upgrade' event (the exact bin wiring)
const server = http.createServer((req, res) => { res.writeHead(404); res.end() })
const wss = new bareWs.Server({ server })
wss.on('connection', (ws, req) => {
  const u = new URL(req.url, 'http://localhost')
  if (u.pathname !== '/xrpc/com.atproto.label.subscribeLabels') { try { ws.destroy() } catch {} ; return }
  const cursor = Number(u.searchParams.get('cursor')) || 0
  const stop = firehose.serve((bytes) => { try { ws.write(Buffer.from(bytes)) } catch {} }, { cursor })
  ws.on('close', stop); ws.on('error', () => { try { stop() } catch {} })
})
await new Promise((r) => server.listen(PORT, r))

// client subscribes and collects decoded frames off the wire
const got = []
const client = new bareWs.Socket('ws://127.0.0.1:' + PORT + '/xrpc/com.atproto.label.subscribeLabels?cursor=0')
client.on('data', (buf) => { const d = decodeLabelsFrame(buf); if (d) got.push(d) })
client.on('error', () => {})
await sleep(800)

assert(got.length >= 2, `client received the 2 replayed label frames over the wire (got ${got.length})`)
assert(got[0].body.seq === 1 && got[1].body.seq === 2, 'wire frames carry monotonic seqs 1,2')
assert(got[0].body.labels[0].val === 'spam' && got[1].body.labels[0].val === '!hide', 'wire labels in order')
assert(verifyLabel(got[0].body.labels[0], labeler.didKey), 'a wire label VERIFIES against the labeler did:key (end-to-end signed)')

// live-tail over the wire
await labeler.label('at://did:plc:c/app.bsky.feed.post/3', 'nsfw')
await sleep(800)
assert(got.length >= 3 && got[got.length - 1].body.labels[0].val === 'nsfw', 'LIVE over the wire: an appended label is pushed as the next seq')

console.log('\nSUBSCRIBE-LABELS WS OK: real bare-ws client replayed + live-tailed VERIFIED labels off the wire')
try { client.destroy() } catch {}
try { wss.close && wss.close() } catch {}
await new Promise((r) => server.close(r))
try { await labeler.close(); await store.close() } catch {}
try { fs.rmSync(dir, { recursive: true, force: true }) } catch {}
process.exit(0)